ashwinpankaj commented on code in PR #8690:
URL: https://github.com/apache/kafka/pull/8690#discussion_r920869174
##########
clients/src/main/java/org/apache/kafka/clients/producer/RoundRobinPartitioner.java:
##########
@@ -65,12 +65,20 @@ public int partition(String topic, Object key, byte[]
keyBytes, Object value, by
}
private int nextValue(String topic) {
- AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> {
- return new AtomicInteger(0);
- });
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
return counter.getAndIncrement();
}
+ @Override
+ public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+ // After onNewBatch is called, we will call partition() again.
+ // So 'rewind' the counter for this topic.
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
+ counter.getAndDecrement();
Review Comment:
I feel that the fix lies in RecordAccumulator as currently it is always
returning `abortForNewBatch`=true from append() for a partition which does not
have a Deque created.
If a partition does not have a deque ,
[accumulator.getOrCreateDeque()](https://github.com/confluentinc/ce-kafka/blob/609c9e8a14f4689de268a9b06b27b5cc982a51c8/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L677)
simply creates an empty ArrayQueue.
When accumulator tries to append a new record, [tryAppend()
](https://github.com/confluentinc/ce-kafka/blob/609c9e8a14f4689de268a9b06b27b5cc982a51c8/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L264)will
return null since ProducerBatch has not been created.
Here are the sequence of events if key value is not set for record -
1. partitioner.partition() is invoked - partition id for topic is
incremented
1. recordaccumulator.append() is invoked with `abortOnNewBatch` arg is set
to true. Accumulator is unable to append record to a batch it returns
RecordAppendResult with abortForNewBatch set to true.
1. partitioner.onNewBatch() is invoked
1. partitioner.partition() is invoked again - partition id for topic is
incremented
1. recordaccumulator.append() is invoked again with `abortOnNewBatch` arg
is set to false. This time accumulator allocates a new ProducerBatch and
appends the record.
Probable fix:
In accumulator.getOrCreateDeque() in addition to creating a Deque, we
should also initialize an empty ProducerBatch for the topicPartition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
