KAFKA-18618: Improve leader change handling of acknowledgements [1/N] #18672

AndrewJSchofield · 2025-01-22T17:49:55Z

There is a problem in the share consumer where records fetched from node A might be acknowledged with node B, if node B used to be the leader of the topic-partition and A becomes unavailable.

Committer Checklist (excluded from commit message)

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

clolov · 2025-01-23T11:53:23Z

I know this is still marked as a Draft, but how difficult will it be to add a test which tests the edge case which uncovered this? I believe I can follow the reasoning otherwise i.e. you only want to send the acknowledgements to the nodes from which the in-flight records originated.

AndrewJSchofield · 2025-01-23T13:02:43Z

I know this is still marked as a Draft, but how difficult will it be to add a test which tests the edge case which uncovered this? I believe I can follow the reasoning otherwise i.e. you only want to send the acknowledgements to the nodes from which the in-flight records originated.

Not that difficult. Might be a follow-on PR to complete the testing, but we have two pieces of work in progress. First, we are developing tests using MockClient which can pretend to be a cluster of multiple brokers in which we can very deliberately introduce specific changes of leadership to exercise all of the code paths. Then, we are starting to run system tests which roll the cluster while performing a long-running workload. It was a rolling restart system test that showed the problem initially.

clolov · 2025-01-23T14:34:18Z

Awesome! Looking forward to seeing those 😊

apoorvmittal10

Looks mostly good to me. Left some minor comments.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java

apoorvmittal10 · 2025-01-29T09:26:32Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java

     *
     * @return the error code
     */
-    public Errors getAcknowledgeErrorCode() {
-        return acknowledgeErrorCode;
+    public KafkaException getAcknowledgeException() {


We generally avoid get prefix in methods so should it be as like below?

Suggested change

public KafkaException getAcknowledgeException() {

public KafkaException acknowledgeException() {

In public classes, yes, I agree. However, here there are many accessor methods already present and I don't think renaming them all improves things.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java

apoorvmittal10 · 2025-01-29T09:29:38Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/NodeAcknowledgements.java

+        this.acknowledgements = acknowledgements;
+    }
+
+    public int getNodeId() {


Suggested change

public int getNodeId() {

public int nodeId() {

apoorvmittal10 · 2025-01-29T09:29:48Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/NodeAcknowledgements.java

+        return nodeId;
+    }
+
+    public Acknowledgements getAcknowledgements() {


Suggested change

public Acknowledgements getAcknowledgements() {

public Acknowledgements acknowledgements() {

apoorvmittal10 · 2025-01-29T09:30:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/NodeAcknowledgements.java

+/**
+ * This class combines Acknowledgements with the id of the node to use for acknowledging.
+ */
+public class NodeAcknowledgements {


nit: Seems this can be a record class itself.

Actually, I don't think that's allowed in the client. Records were introduced in Java 14 and can be used in the broker which builds with Java 17.

Yeah you are right, client is not yet there.

apoorvmittal10

LGTM, just one query.

apoorvmittal10 · 2025-01-29T12:44:32Z

clients/src/main/java/org/apache/kafka/clients/consumer/AcknowledgementCommitCallback.java

     * <li> {@link WakeupException} if {@link KafkaShareConsumer#wakeup()} is called before or while this function is called
     * <li> {@link InterruptException} if the calling thread is interrupted before or while this function is called
     * <li> {@link KafkaException} for any other unrecoverable errors
     * </ul>
+     * <p>Note that if the exception is a retriable exception, the acknowledgement could not be completed and the


nit: For non-retribale exceptions it will fetch anyways again so I expect we mean that even for retriable exception, the fetch will happen again?

Suggested change

* <p>Note that if the exception is a retriable exception, the acknowledgement could not be completed and the

* <p>Note that if the exception is even a retriable exception, the acknowledgement could not be completed and the

I will reword.

apoorvmittal10 · 2025-01-29T13:03:58Z

...ts/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java

-                    metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
-                    fetchAcknowledgementsInFlight.put(tip, acknowledgementsToSend);
+                Acknowledgements acknowledgementsToSend = null;
+                Map<TopicIdPartition, Acknowledgements> nodeAcknowledgementsMap = fetchAcknowledgementsToSend.get(node.id());


Query: Can there be ever some acknowledgments which can starve in the fetchAcknowledgementsToSend when some nodeId is never received from broker?

I think there will need to be a follow-on PR for this. In a way, you're correct, but I think to maintain the ordering of things, we need to fail some acknowledgements more aggressively from the poll loop. I had been avoiding that, but I think it's inevitable actually.

ShivsundarR · 2025-01-29T14:40:20Z

...ts/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java

-                    if (acknowledgements != null) {
-                        acknowledgementsMapForNode.put(tip, acknowledgements);
+                    NodeAcknowledgements acknowledgements = acknowledgementsMap.get(tip);
+                    if ((acknowledgements != null) && (acknowledgements.nodeId() == node.id())) {


Here, should we add a null check for the actual Acknowledgements class inside NodeAcknowledgements.
Although ideally as we already have a null check for NodeAcknowledgements, the case shouldn't ever arise. Just for it be consistent with other parts of the code.

I will rename the variable acknowledgements to nodeAcknowledgements to make it a bit clearer to read. I don't believe that nodeAcknowledgements.acknowledgements() can be null because the empty case is Acknowledgements.empty() and the fetching code in the application thread only sends the acks to be processed if they are non-empty. I can put a null check in the constructor for NodeAcknowledgements.

ShivsundarR · 2025-01-29T15:12:05Z

...ts/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java

+                    Map<TopicIdPartition, Acknowledgements> nodeAcksFromFetchMap = fetchAcknowledgementsToSend.get(nodeId);
+                    if (nodeAcksFromFetchMap != null) {
+                        nodeAcksFromFetchMap.forEach((tip, acks) -> {
+                            metricsManager.recordAcknowledgementSent(acks.size());


Should we remove the mapping (tip, acks) from nodeAcksFromFetchMap(effectively removing them from fetchAcknowledgementsToSend) so that we do not process these acknowledgements again in the next iteration.

Yes, good point.

ShivsundarR · 2025-01-29T15:20:27Z

...ts/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java

+                    Acknowledgements acknowledgements = Acknowledgements.empty();
+                    Map<TopicIdPartition, Acknowledgements> nodeAcksFromFetchMap = fetchAcknowledgementsToSend.get(nodeId);
+                    if (nodeAcksFromFetchMap != null) {
+                        Acknowledgements acksFromFetchMap = nodeAcksFromFetchMap.get(tip);


Should we do a nodeAcksFromFetchMap.remove(tip) here? Although here as we are closing, we would not be using fetchAcknowledgementsToSend again ideally.

I'm not sure it matters because this is on the close path, but I think you're right that it's an improvement.

ShivsundarR

@AndrewJSchofield Thanks for the PR!! Just had a couple of comments.

ShivsundarR

Thanks for the PR! LGTM!

KAFKA-18618: Improve leader change handling of acknowledgements

27d4248

AndrewJSchofield marked this pull request as draft January 22, 2025 17:50

github-actions bot added consumer clients labels Jan 22, 2025

AndrewJSchofield added the KIP-932 Queues for Kafka label Jan 22, 2025

AndrewJSchofield added 3 commits January 24, 2025 12:00

Merge remote-tracking branch 'upstream/trunk' into KAFKA-18618

4683a8f

Twist

38fe5ab

Tidy up forgetting partitions and add tests

bcb865b

AndrewJSchofield marked this pull request as ready for review January 26, 2025 21:31

Add test for disconnection during ack

1ea554a

apoorvmittal10 reviewed Jan 29, 2025

View reviewed changes

AndrewJSchofield added 2 commits January 29, 2025 10:03

Review comments

25840be

Review comments

439fcb5

AndrewJSchofield requested a review from apoorvmittal10 January 29, 2025 10:22

apoorvmittal10 approved these changes Jan 29, 2025

View reviewed changes

Improve comment

88b2b0c

apoorvmittal10 approved these changes Jan 29, 2025

View reviewed changes

ShivsundarR reviewed Jan 29, 2025

View reviewed changes

More review comments

37a48c3

ShivsundarR approved these changes Jan 29, 2025

View reviewed changes

AndrewJSchofield changed the title ~~KAFKA-18618: Improve leader change handling of acknowledgements~~ KAFKA-18618: Improve leader change handling of acknowledgements [1/N] Jan 30, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18618: Improve leader change handling of acknowledgements [1/N] #18672

KAFKA-18618: Improve leader change handling of acknowledgements [1/N] #18672

AndrewJSchofield commented Jan 22, 2025

clolov commented Jan 23, 2025 •

edited

Loading

AndrewJSchofield commented Jan 23, 2025

clolov commented Jan 23, 2025

apoorvmittal10 left a comment

apoorvmittal10 Jan 29, 2025

AndrewJSchofield Jan 29, 2025

apoorvmittal10 Jan 29, 2025

apoorvmittal10 Jan 29, 2025

apoorvmittal10 Jan 29, 2025

AndrewJSchofield Jan 29, 2025

apoorvmittal10 Jan 29, 2025

apoorvmittal10 left a comment

apoorvmittal10 Jan 29, 2025

AndrewJSchofield Jan 29, 2025

apoorvmittal10 Jan 29, 2025

AndrewJSchofield Jan 29, 2025

apoorvmittal10 Jan 29, 2025

ShivsundarR Jan 29, 2025 •

edited

Loading

AndrewJSchofield Jan 29, 2025

ShivsundarR Jan 29, 2025 •

edited

Loading

AndrewJSchofield Jan 29, 2025

ShivsundarR Jan 29, 2025

AndrewJSchofield Jan 29, 2025

ShivsundarR left a comment

ShivsundarR left a comment

	public KafkaException getAcknowledgeException() {
	public KafkaException acknowledgeException() {

	public Acknowledgements getAcknowledgements() {
	public Acknowledgements acknowledgements() {

	* <p>Note that if the exception is a retriable exception, the acknowledgement could not be completed and the
	* <p>Note that if the exception is even a retriable exception, the acknowledgement could not be completed and the

KAFKA-18618: Improve leader change handling of acknowledgements [1/N] #18672

Are you sure you want to change the base?

KAFKA-18618: Improve leader change handling of acknowledgements [1/N] #18672

Conversation

AndrewJSchofield commented Jan 22, 2025

Committer Checklist (excluded from commit message)

clolov commented Jan 23, 2025 • edited Loading

AndrewJSchofield commented Jan 23, 2025

clolov commented Jan 23, 2025

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ShivsundarR Jan 29, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ShivsundarR Jan 29, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ShivsundarR left a comment

Choose a reason for hiding this comment

ShivsundarR left a comment

Choose a reason for hiding this comment

clolov commented Jan 23, 2025 •

edited

Loading

ShivsundarR Jan 29, 2025 •

edited

Loading

ShivsundarR Jan 29, 2025 •

edited

Loading