You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/07/15 21:42:37 UTC
[GitHub] [kafka] artemlivshits commented on a diff in pull request #12365: KAFKA-14020: Performance regression in Producer
artemlivshits commented on code in PR #12365:
URL: https://github.com/apache/kafka/pull/12365#discussion_r922530070
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1465,13 +1465,17 @@ public boolean isDone() {
private class AppendCallbacks<K, V> implements RecordAccumulator.AppendCallbacks {
private final Callback userCallback;
private final ProducerInterceptors<K, V> interceptors;
- private final ProducerRecord<K, V> record;
+ private ProducerRecord<K, V> record;
+ private final String topic;
protected int partition = RecordMetadata.UNKNOWN_PARTITION;
private AppendCallbacks(Callback userCallback, ProducerInterceptors<K, V> interceptors, ProducerRecord<K, V> record) {
this.userCallback = userCallback;
this.interceptors = interceptors;
this.record = record;
+ // Note a record would be null only if the client application has a bug, but we don't want to
+ // have NPE here, because the interceptors would not be notified (see .doSend).
+ topic = record != null ? record.topic() : null;
Review Comment:
https://github.com/apache/kafka/blob/trunk/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala#L1041 has a test, which effectively codifies the contract. I would agree that it's weird to have contract about null handling, but at this point I'd rather preserve whatever behavior is codified.
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1465,13 +1465,15 @@ public boolean isDone() {
private class AppendCallbacks<K, V> implements RecordAccumulator.AppendCallbacks {
private final Callback userCallback;
private final ProducerInterceptors<K, V> interceptors;
- private final ProducerRecord<K, V> record;
+ private ProducerRecord<K, V> record;
Review Comment:
Will do. And partition then should be volatile as well, then.
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1491,20 +1495,25 @@ public void setPartition(int partition) {
if (log.isTraceEnabled()) {
Review Comment:
The intention of an assert is to run in tests, but be disabled in prod, so if my understanding is correct, this is the proper usage.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -297,7 +297,12 @@ public RecordAppendResult append(String topic,
byte maxUsableMagic = apiVersions.maxUsableProduceMagic();
int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key, value, headers));
log.trace("Allocating a new {} byte message buffer for topic {} partition {} with remaining timeout {}ms", size, topic, partition, maxTimeToBlock);
+ // This call may block if we exhausted buffer space.
buffer = free.allocate(size, maxTimeToBlock);
+ // Update the current time in case the buffer allocation blocked above.
+ // NOTE: getting time may be expensive, so calling it under a lock
+ // should be avoided.
+ nowMs = time.milliseconds();
Review Comment:
I synced up with Jun offline, in this change it makes sense to preserve the current behavior (too close to release).
##########
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:
The topicPartition is called once in success case (maybe twice in error case). I'll add a comment.
##########
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;
Review Comment:
The method overrides a callback that is called setPartition to reflect what the caller does with it (it sets partition). I agree that it's a little non-intuitive to do a state transition here but there doesn't seem to be a better place to do it if we want to preserve the behavior -- we need record until here to do the tracing and we cannot do tracing earlier because we may not know the partition; at the same time, we don't want to keep it longer.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org