chia7712 commented on code in PR #20170:
URL: https://github.com/apache/kafka/pull/20170#discussion_r2283468164
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1124,12 +1124,30 @@ private Future<RecordMetadata> doSend(ProducerRecord<K,
V> record, Callback call
ensureValidRecordSize(serializedSize);
long timestamp = record.timestamp() == null ? nowMs :
record.timestamp();
+ // A custom RoundRobinPartitioner may take advantage on the
onNewBatch callback.
+ boolean abortOnNewBatch = false;
+ if (partitionerPlugin.get() instanceof RoundRobinPartitioner) {
+ abortOnNewBatch = true;
+ }
+
// Append the record to the accumulator. Note, that the actual
partition may be
// calculated there and can be accessed via
appendCallbacks.topicPartition.
RecordAccumulator.RecordAppendResult result =
accumulator.append(record.topic(), partition, timestamp, serializedKey,
- serializedValue, headers, appendCallbacks,
remainingWaitMs, nowMs, cluster);
+ serializedValue, headers, appendCallbacks,
remainingWaitMs, nowMs, cluster, abortOnNewBatch);
assert appendCallbacks.getPartition() !=
RecordMetadata.UNKNOWN_PARTITION;
+ // Notify the RoundRobinPartitioner that the previous batch is
full, and request it to return prevPartition to the idle queue.
+ if (result.abortOnNewBatch) {
+ int prevPartition = partition;
+ ((RoundRobinPartitioner)
partitionerPlugin.get()).onNewBatch(record.topic(), cluster, prevPartition);
Review Comment:
That makes sense. We should fix it in the 3.9 branch only
--
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
