You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "rich-c-shop (via GitHub)" <gi...@apache.org> on 2024/09/05 00:28:47 UTC
[PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
rich-c-shop opened a new pull request, #17099:
URL: https://github.com/apache/kafka/pull/17099
1st part of https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=125310116
https://issues.apache.org/jira/browse/KAFKA-8830
2nd part about `onCompletion` changes have larger blast radius, will discuss in different PR
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1747054899
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
Review Comment:
thanks, applied at 16365e4ea526ec9354ddebe2fd4041d78ef3c77b
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1747052310
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,15 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (headers == null) {
+ return;
+ }
+ if (!(headers instanceof RecordHeaders)) {
+ return;
+ }
+ RecordHeaders recordHeaders = (RecordHeaders) headers;
Review Comment:
https://github.com/apache/kafka/commit/988c09f48135a4dc1ab7503cf6ed45ea115063fa
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901967629
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
that works: c9273632266a79c18022a6b7b9858b898a3e0fca
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1750382190
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
Review Comment:
Thanks, 2eb05a7acc
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #17099:
URL: https://github.com/apache/kafka/pull/17099
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1827065687
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1552,6 +1552,7 @@ private class AppendCallbacks implements RecordAccumulator.AppendCallbacks {
private final String recordLogString;
private volatile int partition = RecordMetadata.UNKNOWN_PARTITION;
private volatile TopicPartition topicPartition;
+ private Headers headers;
Review Comment:
please add `final`
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders) {
+ // make a copy of the headers to make sure it is read-only
+ RecordHeaders recordHeaders = (RecordHeaders) headers;
+ headers = new RecordHeaders(recordHeaders);
+ ((RecordHeaders) headers).setReadOnly();
+ }
if (record == null && interceptTopicPartition == null) {
- interceptor.onAcknowledgement(null, exception);
+ interceptor.onAcknowledgement(null, exception, headers);
Review Comment:
We must call `onAcknowledgement(RecordMetadata, Exception, Headers)` even if headers is null. Otherwise, users who implement only the new method will miss this event.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders) {
+ // make a copy of the headers to make sure it is read-only
Review Comment:
we don't need to copy it if it is read-only, right?
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent. It includes the partition and offset of the record.
+ * If an error occurred, the metadata will contain only the topic and possibly the partition.
+
+ * If the partition was not assigned yet due to an error,
+ * it will be set to {@link org.apache.kafka.clients.producer.RecordMetadata#UNKNOWN_PARTITION}.
+ * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during the processing of this record, or null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
+ */
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ onAcknowledgement(metadata, exception);
+ };
Review Comment:
Please remove `;`
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1748346035
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -82,11 +83,12 @@ public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will only contain valid topic and maybe partition.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent
*/
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
- interceptor.onAcknowledgement(metadata, exception);
+ interceptor.onAcknowledgement(metadata, exception, headers);
Review Comment:
I think they should be immutable for send failures too. I can't think of a situation where mutating the headers in the interceptor should work so best to prevent it outright.
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1749469391
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -1099,6 +1104,28 @@ private <T extends Serializer<String>> void doTestHeaders(Class<T> serializerCla
producer.close(Duration.ofMillis(0));
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testHeadersFailure() {
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorForHeaders.class.getName());
+ Serializer<String> keySerializer = mock(Serializer.class);
+ Serializer<String> valueSerializer = mock(Serializer.class);
+
+ KafkaProducer<String, String> producer = new KafkaProducer<>(props, keySerializer, valueSerializer);
+ ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
+ Future<RecordMetadata> future = producer.send(record, (recordMetadata, exception) -> { });
+ try {
+ assertInstanceOf(TimeoutException.class, assertThrows(ExecutionException.class, future::get).getCause());
+ //ensure headers are writable if send failure
Review Comment:
This is header of producer record, different from onAcknowledgement's headers.
Producer record headers is writable because client might want to mutate it before retry. This is same behaviour as today, no change introduced from this PR.
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901267783
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
I agree `ProducerRecord#headers` is never null, however, `ProducerRecord` could be null
It is possible to change
from
```
headers = record != null ? record.headers() : null;
```
to
```
headers = record != null ? record.headers() : new RecordHeaders(null);
```
below. Prefer `new RecordHeaders(null)`?
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L1564
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java#L113
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on PR #17099:
URL: https://github.com/apache/kafka/pull/17099#issuecomment-2369884620
> continuous-integration/jenkins/pr-merge — This commit has test failures
Thanks for reviewing. looks like the test failed in un-related tests. Anything I can help to proceed for the PR to merge?
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901256847
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
Review Comment:
```suggestion
* before partition gets assigned, then partition will be set to {@link RecordMetadata#UNKNOWN_PARTITION}.
```
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901256847
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
Review Comment:
```suggestion
* before partition gets assigned, then partition will be set to {@link RecordMetadata#UNKNOWN_PARTITION}.
```
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1828626827
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent. It includes the partition and offset of the record.
+ * If an error occurred, the metadata will contain only the topic and possibly the partition.
+
+ * If the partition was not assigned yet due to an error,
+ * it will be set to {@link org.apache.kafka.clients.producer.RecordMetadata#UNKNOWN_PARTITION}.
+ * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
Review Comment:
I think the diff of two docs are from rebase. I know update them to same
addressed at e27c61c191
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent. It includes the partition and offset of the record.
+ * If an error occurred, the metadata will contain only the topic and possibly the partition.
+
+ * If the partition was not assigned yet due to an error,
+ * it will be set to {@link org.apache.kafka.clients.producer.RecordMetadata#UNKNOWN_PARTITION}.
+ * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during the processing of this record, or null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
+ */
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ onAcknowledgement(metadata, exception);
+ };
Review Comment:
addressed at e27c61c191
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1827965638
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders) {
Review Comment:
yes, that's a good idea
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901267783
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
I agree `ProducerRecord#headers` is never null, however, `ProducerRecord` could be null
It is possible to change
from
```
headers = record != null ? record.headers() : null;
```
to
```
headers = record != null ? record.headers() : new RecordHeaders(null);
```
below.
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L1564
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java#L113
---
Prefer `new RecordHeaders(null)`❓
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901261096
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders && !((RecordHeaders) headers).isReadOnly()) {
+ // make a copy of the headers to make sure it is read-only
Review Comment:
makes sense, 235e53f987
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1905466781
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1546,6 +1547,7 @@ private AppendCallbacks(Callback userCallback, ProducerInterceptors<K, V> interc
// whole lifetime of the batch.
// We don't want to have an NPE here, because the interceptors would not be notified (see .doSend).
topic = record != null ? record.topic() : null;
+ headers = record != null ? record.headers() : new RecordHeaders();
Review Comment:
To maintain header immutability, especially given your efforts in this area, it might be prudent to pass immutable headers when record is null.
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to {@link RecordMetadata#UNKNOWN_PARTITION}.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
+ * throws an exception.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * This method will generally execute in the background I/O thread, so the implementation should be reasonably fast.
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to {@link RecordMetadata#UNKNOWN_PARTITION}.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent.
Review Comment:
Should we remind users that they should not change the headers?
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1745593973
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
Review Comment:
Thanks for the feedback on docs. This is copied from existing doc, shall we change in both places? To keep minimal change in this PR, what do you think if I modify both in following PR?
e.g.
https://github.com/apache/kafka/blob/f776a81b0fdcf72d36db76c3bfc1fd8f098e9ba3/clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java#L74
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on PR #17099:
URL: https://github.com/apache/kafka/pull/17099#issuecomment-2333956366
> Could you implement two different interceptor classes in ProducerInterceptorsTest, one with the new signature of onAcknowledgement and one with the old, and then run the same parameterized tests to show they are equivalent in behaviour?
sure. I added 2nd interceptor at 5c1c8430a4. I didn't use parameterized tests because the test already include two interceptors, I just change later one to new signature
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1745857998
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,12 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (!(headers instanceof RecordHeaders)) {
Review Comment:
For what cases do we expect this to be "false"? -- given that it is test code, I am wondering if we need it, or if it could (in the worst case) even mask a potential bug?
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901267783
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
I agree `ProducerRecord#headers` is never null, however, `ProducerRecord` could be null
It is possible to change
from
```
headers = record != null ? record.headers() : null;
```
to
```
headers = record != null ? record.headers() : new RecordHeaders(null);
```
below.
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L1564
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java#L113
---
Prefer `new RecordHeaders(null)`❓ To me, I think `new RecordHeaders(null)` is a bit leaking internal implementation outside `ProducerRecord`
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901267783
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
I agree `ProducerRecord#header` is never null, however, `ProducerRecord` could be null
It is possible to change
from
```
headers = record != null ? record.headers() : null;
```
to
```
headers = record != null ? record.headers() : new RecordHeaders(null);
```
below. Prefer that?
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L1564
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java#L113
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on PR #17099:
URL: https://github.com/apache/kafka/pull/17099#issuecomment-2788107901
@chia7712 I address the two minor comments and rebase on latest trunk. Do you have time to take final look to see if we are ok to merge?
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1747034130
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,12 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (!(headers instanceof RecordHeaders)) {
Review Comment:
removed at 988c09f481
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1746375669
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -82,11 +83,12 @@ public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will only contain valid topic and maybe partition.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent
*/
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
- interceptor.onAcknowledgement(metadata, exception);
+ interceptor.onAcknowledgement(metadata, exception, headers);
Review Comment:
I know this came up during the KIP review, but really the headers ought to be read-only here. You could make a read-only implementation of `Headers` that delegates all read operations to the actual `Headers`. Otherwise, some cunning person might start modifying the headers structure and passing it between interceptors.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -82,11 +83,12 @@ public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will only contain valid topic and maybe partition.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent
*/
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
- interceptor.onAcknowledgement(metadata, exception);
+ interceptor.onAcknowledgement(metadata, exception, headers);
Review Comment:
Ah, no. Not necessary. The producer has already made them read-only. :)
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,15 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (headers == null) {
+ return;
+ }
+ if (!(headers instanceof RecordHeaders)) {
+ return;
+ }
+ RecordHeaders recordHeaders = (RecordHeaders) headers;
Review Comment:
Well, I think it is important that they are read-only so this assertion probably does have some value.
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,12 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (!(headers instanceof RecordHeaders)) {
Review Comment:
Yes, I agree.
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1827956965
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
Review Comment:
you are 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1827966644
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders) {
+ // make a copy of the headers to make sure it is read-only
Review Comment:
yes, you are right, will improve as https://github.com/apache/kafka/pull/17099/files#r1784802691 mention
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1828626528
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1552,6 +1552,7 @@ private class AppendCallbacks implements RecordAccumulator.AppendCallbacks {
private final String recordLogString;
private volatile int partition = RecordMetadata.UNKNOWN_PARTITION;
private volatile TopicPartition topicPartition;
+ private Headers headers;
Review Comment:
addressed at e27c61c191
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
Review Comment:
addressed at e27c61c191
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1828632310
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent. It includes the partition and offset of the record.
+ * If an error occurred, the metadata will contain only the topic and possibly the partition.
+
+ * If the partition was not assigned yet due to an error,
+ * it will be set to {@link org.apache.kafka.clients.producer.RecordMetadata#UNKNOWN_PARTITION}.
+ * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during the processing of this record, or null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
is it aligned now?
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent. It includes the partition and offset of the record.
+ * If an error occurred, the metadata will contain only the topic and possibly the partition.
+
+ * If the partition was not assigned yet due to an error,
+ * it will be set to {@link org.apache.kafka.clients.producer.RecordMetadata#UNKNOWN_PARTITION}.
+ * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during the processing of this record, or null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
thanks, is it aligned now?
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1747027687
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,12 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (!(headers instanceof RecordHeaders)) {
Review Comment:
Good call, I will remove it
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r2033443850
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1546,6 +1547,7 @@ private AppendCallbacks(Callback userCallback, ProducerInterceptors<K, V> interc
// whole lifetime of the batch.
// We don't want to have an NPE here, because the interceptors would not be notified (see .doSend).
topic = record != null ? record.topic() : null;
+ headers = record != null ? record.headers() : new RecordHeaders();
Review Comment:
fca85102d87580a593098b8c1eb5ee97d941a43f
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to {@link RecordMetadata#UNKNOWN_PARTITION}.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
+ * throws an exception.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * This method will generally execute in the background I/O thread, so the implementation should be reasonably fast.
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to {@link RecordMetadata#UNKNOWN_PARTITION}.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent.
Review Comment:
fca85102d87580a593098b8c1eb5ee97d941a43f
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1784733672
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent. It includes the partition and offset of the record.
+ * If an error occurred, the metadata will contain only the topic and possibly the partition.
+
+ * If the partition was not assigned yet due to an error,
+ * it will be set to {@link org.apache.kafka.clients.producer.RecordMetadata#UNKNOWN_PARTITION}.
+ * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during the processing of this record, or null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
format seems off here (lines not aligned)
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1745852320
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
Review Comment:
I think it's fine to apply this suggestion to the existing comment in this PR right away. If you prefer to do a separate PR, also ok.
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1749467047
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -82,11 +83,12 @@ public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will only contain valid topic and maybe partition.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent
*/
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
- interceptor.onAcknowledgement(metadata, exception);
+ interceptor.onAcknowledgement(metadata, exception, headers);
Review Comment:
fair enough. I just add `testHeadersFailure` to make sure onAcknowledgement headers is read only
7bd95cf30736a13ac74f6ac2633824b29e49ad33
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17099:
URL: https://github.com/apache/kafka/pull/17099#issuecomment-2785142757
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on PR #17099:
URL: https://github.com/apache/kafka/pull/17099#issuecomment-2786823125
> This branch has conflicts that must be resolved
I am going to rebase
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901261665
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
Review Comment:
great suggestion, 235e53f987
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
Review Comment:
235e53f987
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901261512
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
Review Comment:
yes, updated to align with previous one. It was coping old one before rebase.
235e53f987
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1746993468
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,15 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (headers == null) {
+ return;
+ }
+ if (!(headers instanceof RecordHeaders)) {
+ return;
+ }
+ RecordHeaders recordHeaders = (RecordHeaders) headers;
Review Comment:
Yes, the goal is to verify it is read only
Turn out this code is not hit during test. It is reference in one test, but the test does not send any record. I will fix it.
https://github.com/apache/kafka/blob/6c582bd2c5644bdbe2b92f6a50547ed27ae3908a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java#L2061-L2075
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1750312577
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,37 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
Review Comment:
nit: Formatting of the javadoc comment has gone a bit awry.
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1828631651
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -1099,6 +1104,28 @@ private <T extends Serializer<String>> void doTestHeaders(Class<T> serializerCla
producer.close(Duration.ofMillis(0));
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testHeadersFailure() {
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorForHeaders.class.getName());
+ Serializer<String> keySerializer = mock(Serializer.class);
+ Serializer<String> valueSerializer = mock(Serializer.class);
+
+ KafkaProducer<String, String> producer = new KafkaProducer<>(props, keySerializer, valueSerializer);
+ ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
+ Future<RecordMetadata> future = producer.send(record, (recordMetadata, exception) -> { });
+ try {
+ assertInstanceOf(TimeoutException.class, assertThrows(ExecutionException.class, future::get).getCause());
Review Comment:
yes, the test was slow and timeout after 1 min. I just add MAX_BLOCK_MS=500 like other tests, it finish within 1s
f2438262b4
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -1099,6 +1104,28 @@ private <T extends Serializer<String>> void doTestHeaders(Class<T> serializerCla
producer.close(Duration.ofMillis(0));
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testHeadersFailure() {
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorForHeaders.class.getName());
+ Serializer<String> keySerializer = mock(Serializer.class);
+ Serializer<String> valueSerializer = mock(Serializer.class);
+
+ KafkaProducer<String, String> producer = new KafkaProducer<>(props, keySerializer, valueSerializer);
+ ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
+ Future<RecordMetadata> future = producer.send(record, (recordMetadata, exception) -> { });
+ try {
+ assertInstanceOf(TimeoutException.class, assertThrows(ExecutionException.class, future::get).getCause());
Review Comment:
addressed at f2438262b4
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on PR #17099:
URL: https://github.com/apache/kafka/pull/17099#issuecomment-2787762389
> Run './gradlew :clients:spotlessApply' to fix these violations
run it locally gives no error. I will follow CI result to fix the import ordering then.
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901858197
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
I see, but what about
```
headers = record != null ? record.headers() : new RecordHeaders();
```
Just to pass empty headers obj instead of null to the callback, to make it's implementation simpler (avoid null checks/NPE). WDYT?
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on PR #17099:
URL: https://github.com/apache/kafka/pull/17099#issuecomment-2569525607
squash and rebased
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901267783
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
I agree `ProducerRecord#headers` is never null, however, `ProducerRecord` could be null
It is possible to change
from
```
headers = record != null ? record.headers() : null;
```
to
```
headers = record != null ? record.headers() : new RecordHeaders(null);
```
below. Prefer that?
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L1564
https://github.com/apache/kafka/blob/235e53f987cd5cd8841b8ba1d46bacd2be6955e1/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java#L113
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1830339037
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders && !((RecordHeaders) headers).isReadOnly()) {
+ // make a copy of the headers to make sure it is read-only
Review Comment:
how about 339e3bad72?
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1745665159
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,15 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (headers == null) {
+ return;
+ }
+ if (!(headers instanceof RecordHeaders)) {
Review Comment:
sure, fixed at 6c582bd2c5
turn out `headers==null` is not need, it is handled as part of `!(headers instanceof RecordHeaders)`
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1745172743
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
Review Comment:
```suggestion
* This method is called just before the user callback is invoked, or in cases when
* <code>KafkaProducer.send()</code> throws an exception.
```
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
+ * throws an exception.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * This method will generally execute in the background I/O thread, so the implementation should be reasonably fast.
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
Review Comment:
```suggestion
* If an error occurred, the metadata will contain only the topic and possibly the partition.
```
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
+ * throws an exception.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * This method will generally execute in the background I/O thread, so the implementation should be reasonably fast.
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
Review Comment:
```suggestion
* @param metadata The metadata for the record that was sent. It includes the partition and offset of the record.
```
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
+ * throws an exception.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * This method will generally execute in the background I/O thread, so the implementation should be reasonably fast.
Review Comment:
```suggestion
* The implementation of this method should be fast as it generally executes in a background
* I/O thread. A slow implementation could delay the sending of messages from other threads.
```
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,15 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (headers == null) {
+ return;
+ }
+ if (!(headers instanceof RecordHeaders)) {
+ return;
+ }
+ RecordHeaders recordHeaders = (RecordHeaders) headers;
Review Comment:
I can't see any test setting header as `RecordHeaders` so am not sure what the benefit of this assertion if there is no actual test for it.
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
+ * throws an exception.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * This method will generally execute in the background I/O thread, so the implementation should be reasonably fast.
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent
Review Comment:
```suggestion
* @param headers The headers for the record that was sent. This parameter may be null.
```
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
+ * throws an exception.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
Review Comment:
This can be moved to the end of the Java doc and just before the params definition as a note
```suggestion
* Note that any exception thrown by this method will be ignored by the caller.
```
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -2397,7 +2399,15 @@ public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> reco
}
@Override
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
+ if (headers == null) {
+ return;
+ }
+ if (!(headers instanceof RecordHeaders)) {
Review Comment:
maybe merge these 2 `If`s instead of having 2 of them
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
+ * throws an exception.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * This method will generally execute in the background I/O thread, so the implementation should be reasonably fast.
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
Review Comment:
```suggestion
* If the partition was not assigned yet due to an error,
* it will be set to {@link org.apache.kafka.clients.producer.RecordMetadata#NO_PARTITION}.
```
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -86,7 +87,32 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {};
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is generally called just before the user callback is called, and in additional cases when <code>KafkaProducer.send()</code>
+ * throws an exception.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * This method will generally execute in the background I/O thread, so the implementation should be reasonably fast.
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
Review Comment:
```suggestion
* @param exception The exception thrown during the processing of this record, or null if no error occurred.
```
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1747036431
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -82,11 +83,12 @@ public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will only contain valid topic and maybe partition.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent
*/
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
- interceptor.onAcknowledgement(metadata, exception);
+ interceptor.onAcknowledgement(metadata, exception, headers);
Review Comment:
Actually, when send successfully, headers are read only. But when send failure, they are NOT. I think it is reasonable to be mutable when send failure, what do you think ❓
I have a test case for it now, 988c09f481
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901275510
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -1099,6 +1105,29 @@ private <T extends Serializer<String>> void doTestHeaders(Class<T> serializerCla
producer.close(Duration.ofMillis(0));
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testHeadersFailure() {
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 500);
Review Comment:
sure, dd79971e3d
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -1099,6 +1105,29 @@ private <T extends Serializer<String>> void doTestHeaders(Class<T> serializerCla
producer.close(Duration.ofMillis(0));
}
+ @SuppressWarnings("unchecked")
Review Comment:
yes, that was the reason. Thanks for the suggestion
dd79971e3d
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901858197
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
I see, but what about
```
headers = record != null ? record.headers() : new RecordHeaders();
```
Just to pass empty headers obj instead of null to the callback, to make it's implementation simpler (avoid null checks/NPE)
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1901095800
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
Review Comment:
We have kind of a duplicate here
```
A slow implementation could delay the sending of messages from other threads.
```
and
```
Otherwise, sending of messages from other threads could be delayed.
```
I would maybe keep the latter just to keep the same we have in the other `onAcknowledgement`
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -1099,6 +1105,29 @@ private <T extends Serializer<String>> void doTestHeaders(Class<T> serializerCla
producer.close(Duration.ofMillis(0));
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testHeadersFailure() {
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 500);
Review Comment:
I would push this config even lower (5 maybe?). This will make the test block uselessly for 500 ms on `waitOnMetadata`, to then fail (which is truly the bit this test is after, so let's get there asap and avoid making the test suite slower)
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
Review Comment:
what about using a link to avoid this falling behind again
```suggestion
* before partition gets assigned, then partition will be set to {@link RecordMetadata#UNKNOWN_PARTITION}.
```
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders && !((RecordHeaders) headers).isReadOnly()) {
+ // make a copy of the headers to make sure it is read-only
Review Comment:
I agree with how you extended the comment @rich-c-shop, but still I find we could better clarify what @chia7712 suggested. What about something along the lines of:
```suggestion
// make a copy of the headers to make sure we don't change the state of origin record's headers.
// Original headers are still writable because client might want to mutate them before retrying.
```
(basically we have 2 things here, we don't want to change the record headers within this func, and they also need to be writable)
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
This suggestion makes sense to me. Thoughts @rich-c-shop ?
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
Review Comment:
ditto
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -1099,6 +1105,29 @@ private <T extends Serializer<String>> void doTestHeaders(Class<T> serializerCla
producer.close(Duration.ofMillis(0));
}
+ @SuppressWarnings("unchecked")
Review Comment:
why do we need to suppress here?
If it's only because of the `Serializer<String> keySerializer = mock(Serializer.class);` I would say that we should better fix it to `Serializer<String> keySerializer = mock(StringSerializer.class);` and avoid swallowing the checks. Makes sense?
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r2033386184
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1546,6 +1547,7 @@ private AppendCallbacks(Callback userCallback, ProducerInterceptors<K, V> interc
// whole lifetime of the batch.
// We don't want to have an NPE here, because the interceptors would not be notified (see .doSend).
topic = record != null ? record.topic() : null;
+ headers = record != null ? record.headers() : new RecordHeaders();
Review Comment:
miss this comment.
good point, I will update it
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1828626942
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -82,11 +84,12 @@ public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will only contain valid topic and maybe partition.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent
*/
- public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+ public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
- interceptor.onAcknowledgement(metadata, exception);
+ interceptor.onAcknowledgement(metadata, exception, headers);
Review Comment:
addressed at e27c61c191
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders) {
Review Comment:
addressed at e27c61c191
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders) {
+ // make a copy of the headers to make sure it is read-only
Review Comment:
addressed at e27c61c191
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1863883205
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java:
##########
@@ -81,12 +82,40 @@ public interface ProducerInterceptor<K, V> extends Configurable, AutoCloseable {
* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
* If an error occurred, metadata will contain only valid topic and maybe
* partition. If partition is not given in ProducerRecord and an error occurs
- * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
* The metadata may be null if the client passed null record to
* {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
- void onAcknowledgement(RecordMetadata metadata, Exception exception);
+ default void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
+
+ /**
+ * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
+ * it gets sent to the server.
+ * <p>
+ * This method is called just before the user callback is invoked, or in cases when
+ * <code>KafkaProducer.send()</code> throws an exception.
+ * throws an exception.
+ * <p>
+ * Note that any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * The implementation of this method should be fast as it generally executes in a background
+ * I/O thread. A slow implementation could delay the sending of messages from other threads.
+
+ * Otherwise, sending of messages from other threads could be delayed.
+ *
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.UNKNOWN_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
+ * @param exception The exception thrown during processing of this record. Null if no error occurred.
+ * @param headers The headers for the record that was sent. This parameter may be null.
Review Comment:
Since `ProducerRecord#header` is never null, it would be more appropriate to pass an empty header rather than null. This aligns with the existing behavior and avoids introducing unnecessary null checks. WDYT?
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java#L83
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1830330819
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders && !((RecordHeaders) headers).isReadOnly()) {
+ // make a copy of the headers to make sure it is read-only
Review Comment:
The copy is required because we don't want to change the state of origin record's headers, right? If so, could you please enhance the comment?
--
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
Re: [PR] [KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement [kafka]
Posted by "rich-c-shop (via GitHub)" <gi...@apache.org>.
rich-c-shop commented on code in PR #17099:
URL: https://github.com/apache/kafka/pull/17099#discussion_r1830337731
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java:
##########
@@ -107,14 +110,21 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
try {
+ Headers headers = record != null ? record.headers() : null;
+ if (headers instanceof RecordHeaders && !((RecordHeaders) headers).isReadOnly()) {
+ // make a copy of the headers to make sure it is read-only
Review Comment:
yes, that's correct, https://github.com/apache/kafka/pull/17099#discussion_r1749469391
--
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