junrao commented on code in PR #20170:
URL: https://github.com/apache/kafka/pull/20170#discussion_r2283415017
##########
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:
Hmm, onNewBatch() is already removed in trunk and Partitioner.partition()
should be called only once for each record in the producer. So, the current
RoundRobinPartitioner should work in trunk, right?
--
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
