junrao commented on code in PR #11689:
URL: https://github.com/apache/kafka/pull/11689#discussion_r852429053
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -995,8 +995,17 @@ private Future<RecordMetadata> doSend(ProducerRecord<K, V>
record, Callback call
// for other exceptions throw directly
} catch (ApiException e) {
log.debug("Exception occurred during message send:", e);
- if (callback != null)
- callback.onCompletion(null, e);
+ // producer callback will make sure to call both 'callback' and
interceptor callback
+ if (tp == null) {
+ // set topicPartition to -1 when null
+ tp = ProducerInterceptors.extractTopicPartition(record);
+ }
+
+ Callback interceptCallback = new InterceptorCallback<>(callback,
this.interceptors, tp);
+
+ // The onCompletion callback does expect a non-null metadata, but
one will be created inside
+ // the interceptor's onCompletion implementation before the user's
callback is invoked.
+ interceptCallback.onCompletion(null, e);
Review Comment:
@philipnee : It seems that we may have introduced a bug here.
interceptors.onAcknowledgement() is now called twice, once through
interceptCallback.onCompletion() and another through interceptors.onSendError()
in line 1010 below.
--
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
