[KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement #17099

rich-c-shop · 2024-09-05T00:28:44Z

close https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=125310116
https://issues.apache.org/jira/browse/KAFKA-8830

Two sets of tests are added:

KafkaProducerTest
- when send success, both record.headers() and onAcknowledgement headers are read only
- when send failure, record.headers() is writable as before and onAcknowledgement headers is read only
ProducerInterceptorsTest
- make both old and new onAcknowledgement method are called successfully

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

OmniaGM

Thanks for the PR, I left few comments.

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

AndrewJSchofield

Thanks for the PR.

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?

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

rich-c-shop · 2024-09-06T12:34:42Z

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 5c1c843. I didn't use parameterized tests because the test already include two interceptors, I just change later one to new signature

rich-c-shop · 2024-09-09T02:16:24Z

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

+        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


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.

AndrewJSchofield

Apart from a trivial formatting problem, looks good to me.

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

AndrewJSchofield

Thanks for the PR. lgtm

rich-c-shop · 2024-09-24T00:55:08Z

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?

lianetm

Hello, thanks for the PR! Took a first pass and left some comments for consideration.

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

lianetm · 2024-10-02T15:05:42Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

+     * @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.


format seems off here (lines not aligned)

thanks, is it aligned now?

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

chia7712

@rich-c-shop thanks for this contribution!

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java

chia7712

@rich-c-shop thanks for this patch. one small comment is left.

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java

chia7712 · 2024-11-29T19:29:10Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

+     *                 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.


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 suggestion makes sense to me. Thoughts @rich-c-shop ?

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.

kafka/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

Line 1564 in 235e53f

headers = record != null ? record.headers() : null;

kafka/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java

Line 113 in 235e53f

Headers headers = record != null ? record.headers() : null;

Prefer new RecordHeaders(null)❓ To me, I think new RecordHeaders(null) is a bit leaking internal implementation outside ProducerRecord

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?

that works: c927363

lianetm

Thanks for the updates! Left some minor comments, and there is this one from @chia7712 that hasn't been addressed. Please take a look when you can, this one seems almost ready to me. Thanks!

lianetm · 2025-01-02T16:22:50Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

+     *                 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.


This suggestion makes sense to me. Thoughts @rich-c-shop ?

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

lianetm

Thanks for the updates! Just one follow-up comment, and also could you please merge trunk latest changes?

lianetm · 2025-01-03T14:52:10Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

+     *                 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.


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?

simplify if condition fix test to send record and trigger onAcknowledgement add 2nd ProducerInterceptor for test Apply suggestions from code review Co-authored-by: Omnia Ibrahim <o.g.h.ibrahim@gmail.com> ./gradlew :clients:spotlessApply new testHeadersFailure test to cover onAcknowledgement headers mutability reduce PR diff fix javadoc format address comments with minor doc change and headers isReadOnly checking set MAX_BLOCK_MS_CONFIG to timout fast in test add comment about why copy headers address comments shortern max block of test and fix unchecked use empty header

rich-c-shop · 2025-01-03T16:46:59Z

squash and rebased

lianetm

Thanks for the updates @rich-c-shop , LGMT.
@chia7712 and @AndrewJSchofield let me know if you have any other comments, otherwise I will carry on and merge.

Failed test EagerConsumerCoordinatorTest.testOutdatedCoordinatorAssignment seems unrelated to me, and it's already tracked as failing in trunk with https://issues.apache.org/jira/browse/KAFKA-15900.

chia7712

@rich-c-shop thanks for this PR and sorry for late reviews. two minor comments are left. PTAL

chia7712 · 2025-01-07T13:37:39Z

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();


To maintain header immutability, especially given your efforts in this area, it might be prudent to pass immutable headers when record is null.

chia7712 · 2025-01-07T13:38:21Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java

+     *                 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.


Should we remind users that they should not change the headers?

rich-c-shop mentioned this pull request Sep 5, 2024

[KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement and onComplete #17079

Closed

3 tasks

rich-c-shop marked this pull request as ready for review September 5, 2024 02:48

OmniaGM added the producer label Sep 5, 2024

OmniaGM reviewed Sep 5, 2024

View reviewed changes

mjsax reviewed Sep 5, 2024

View reviewed changes

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java Outdated Show resolved Hide resolved

AndrewJSchofield requested changes Sep 6, 2024

View reviewed changes

rich-c-shop requested a review from AndrewJSchofield September 6, 2024 13:57

rich-c-shop commented Sep 9, 2024

View reviewed changes

AndrewJSchofield requested changes Sep 9, 2024

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java Outdated Show resolved Hide resolved

AndrewJSchofield approved these changes Sep 9, 2024

View reviewed changes

mjsax added the kip Requires or implements a KIP label Sep 28, 2024

lianetm reviewed Oct 2, 2024

View reviewed changes

chia7712 reviewed Nov 3, 2024

View reviewed changes

github-actions bot added the clients label Nov 5, 2024

rich-c-shop requested review from lianetm and chia7712 November 5, 2024 02:11

chia7712 added the ci-approved label Nov 6, 2024

chia7712 reviewed Nov 6, 2024

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java Outdated Show resolved Hide resolved

chia7712 reviewed Nov 29, 2024

View reviewed changes

lianetm reviewed Jan 2, 2025

View reviewed changes

lianetm reviewed Jan 3, 2025

View reviewed changes

rich-c-shop force-pushed the KIP-512-add-headers-to-onAcknowledgement branch from c927363 to 02298d5 Compare January 3, 2025 16:46

lianetm approved these changes Jan 6, 2025

View reviewed changes

chia7712 reviewed Jan 7, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement #17099

[KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement #17099

rich-c-shop commented Sep 5, 2024 •

edited

Loading

OmniaGM left a comment

AndrewJSchofield left a comment

rich-c-shop commented Sep 6, 2024

rich-c-shop Sep 9, 2024

AndrewJSchofield left a comment

AndrewJSchofield left a comment

rich-c-shop commented Sep 24, 2024

lianetm left a comment

lianetm Oct 2, 2024 •

edited

Loading

rich-c-shop Nov 5, 2024 •

edited

Loading

chia7712 left a comment

chia7712 left a comment

chia7712 Nov 29, 2024

lianetm Jan 2, 2025

rich-c-shop Jan 2, 2025 •

edited

Loading

lianetm Jan 3, 2025 •

edited

Loading

rich-c-shop Jan 3, 2025

lianetm left a comment

lianetm Jan 2, 2025

lianetm left a comment

lianetm Jan 3, 2025 •

edited

Loading

rich-c-shop commented Jan 3, 2025

lianetm left a comment

chia7712 left a comment

chia7712 Jan 7, 2025

chia7712 Jan 7, 2025

[KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement #17099

Are you sure you want to change the base?

[KAFKA-8830] KIP-512: make Record Headers available in onAcknowledgement #17099

Conversation

rich-c-shop commented Sep 5, 2024 • edited Loading

Committer Checklist (excluded from commit message)

OmniaGM left a comment

Choose a reason for hiding this comment

AndrewJSchofield left a comment

Choose a reason for hiding this comment

rich-c-shop commented Sep 6, 2024

rich-c-shop Sep 9, 2024

Choose a reason for hiding this comment

AndrewJSchofield left a comment

Choose a reason for hiding this comment

AndrewJSchofield left a comment

Choose a reason for hiding this comment

rich-c-shop commented Sep 24, 2024

lianetm left a comment

Choose a reason for hiding this comment

lianetm Oct 2, 2024 • edited Loading

Choose a reason for hiding this comment

rich-c-shop Nov 5, 2024 • edited Loading

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

chia7712 Nov 29, 2024

Choose a reason for hiding this comment

lianetm Jan 2, 2025

Choose a reason for hiding this comment

rich-c-shop Jan 2, 2025 • edited Loading

Choose a reason for hiding this comment

lianetm Jan 3, 2025 • edited Loading

Choose a reason for hiding this comment

rich-c-shop Jan 3, 2025

Choose a reason for hiding this comment

lianetm left a comment

Choose a reason for hiding this comment

lianetm Jan 2, 2025

Choose a reason for hiding this comment

lianetm left a comment

Choose a reason for hiding this comment

lianetm Jan 3, 2025 • edited Loading

Choose a reason for hiding this comment

rich-c-shop commented Jan 3, 2025

lianetm left a comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

chia7712 Jan 7, 2025

Choose a reason for hiding this comment

chia7712 Jan 7, 2025

Choose a reason for hiding this comment

rich-c-shop commented Sep 5, 2024 •

edited

Loading

lianetm Oct 2, 2024 •

edited

Loading

rich-c-shop Nov 5, 2024 •

edited

Loading

rich-c-shop Jan 2, 2025 •

edited

Loading

lianetm Jan 3, 2025 •

edited

Loading

lianetm Jan 3, 2025 •

edited

Loading