chia7712 commented on code in PR #20170:
URL: https://github.com/apache/kafka/pull/20170#discussion_r2283680413
##########
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 is an interesting story
#17620 was not backported to 3.9, and the fix in 4.0 was removed along with
the deprecated method. In other words, the fix never made it into any release
...
I will backport #17620 to 3.9
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
