ijuma commented on code in PR #12365:
URL: https://github.com/apache/kafka/pull/12365#discussion_r922782270
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1491,20 +1495,25 @@ public void setPartition(int partition) {
if (log.isTraceEnabled()) {
// Log the message here, because we don't know the partition
before that.
- log.trace("Attempting to append record {} with callback {} to
topic {} partition {}", record, userCallback, record.topic(), partition);
+ log.trace("Attempting to append record {} with callback {} to
topic {} partition {}", record, userCallback, topic, partition);
}
+
+ // Reset record to null here so that it doesn't have to be alive
as long as the batch is.
+ record = null;
}
public int getPartition() {
return partition;
}
public TopicPartition topicPartition() {
- if (record == null)
- return null;
- return partition == RecordMetadata.UNKNOWN_PARTITION
- ? ProducerInterceptors.extractTopicPartition(record)
- : new TopicPartition(record.topic(), partition);
+ if (partition != RecordMetadata.UNKNOWN_PARTITION)
+ return new TopicPartition(topic, partition);
Review Comment:
Even so, the way this method is used can change over time. And then you end
up with a lot of unexpected allocation. The way I suggested is more robust.
--
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
