fix deleteRecord #1425
fix deleteRecord #1425
Conversation
| @@ -81,6 +81,28 @@ func (err ConfigurationError) Error() string { | |||
| // See https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes | |||
| type KError int16 | |||
|
|
|||
| // MultiError is used to contain multi error. | |||
| type MultiError struct { | |||
There was a problem hiding this comment.
What's the reason to add this?
There was a problem hiding this comment.
Currently the delete record request is sent to leader by leader. Thus we may receive multi error response and probably the errors are not the same. I use MultiError to contain the array of error.
| return err | ||
| partitionPerBroker := make(map[*Broker][]int32) | ||
| for partition := range partitionOffsets { | ||
| broker, err := ca.client.Leader(topic, partition) |
There was a problem hiding this comment.
formatting looks strange, can you please run gofmt again?
There was a problem hiding this comment.
gofmt again but nothing changed. Can you please tell me what's the format problem. I'll fix it manually.
|
Thanks! What @varun06 said; apart from that could you please squash your commits 🙏 |
I have rebased the commit & push again. But I'm not sure whether it is ok to do it right now and shall I squash the commit again after the code review finished. |
The pr is to fix a bug about delete records. Currently sarama send the request to the controller. It should be sent to the leader according the code in kafka client.
related with #1377