(kafka) branch 4.0 updated: KAFKA-18476: KafkaStreams should swallow TransactionAbortedException (#18487)

[mj...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mjsax pushed a commit to branch 4.0
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/4.0 by this push:
     new 6a9ba19a5f7 KAFKA-18476: KafkaStreams should swallow TransactionAbortedException (#18487)
6a9ba19a5f7 is described below

commit 6a9ba19a5f716e09cf6385e404fb304e807609b3
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Tue Jan 14 18:03:14 2025 -0800

    KAFKA-18476: KafkaStreams should swallow TransactionAbortedException (#18487)
    
    TransactionAbortedException is a follow up error to a previous error,
    and such a previous error would already be handled when
    `producer.abortTransaction()` is called. Thus, a
    TransactionAbortedException can just be silently swallowed.
    
    Reviewers: Bill Bejeck <bi...@confluent.io>
---
 .../processor/internals/RecordCollectorImpl.java   |  6 +++++
 .../processor/internals/RecordCollectorTest.java   | 26 ++++++++++++++++++++++
 2 files changed, 32 insertions(+)

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
index 81db581d04f..edc67083f10 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
@@ -33,6 +33,7 @@ import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.errors.SecurityDisabledException;
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.errors.TransactionAbortedException;
 import org.apache.kafka.common.errors.UnknownServerException;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
@@ -441,6 +442,11 @@ public class RecordCollectorImpl implements RecordCollector {
             errorMessage += "\nWritten offsets would not be recorded and no more records would be sent since the producer is fenced, " +
                 "indicating the task may be migrated out";
             sendException.set(new TaskMigratedException(errorMessage, productionException));
+        } else if (productionException instanceof TransactionAbortedException) {
+            // swallow silently
+            //
+            // TransactionAbortedException is only thrown after `abortTransaction()` was called,
+            // so it's only a followup error, and Kafka Streams is already handling the original error
         } else {
             final ProductionExceptionHandlerResponse response;
             try {
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
index e68d5a4396f..b01b87ed85f 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
@@ -33,6 +33,7 @@ import org.apache.kafka.common.errors.InvalidProducerEpochException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.errors.TransactionAbortedException;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
@@ -1806,6 +1807,31 @@ public class RecordCollectorTest {
         }
     }
 
+    @Test
+    public void shouldSwallowTransactionAbortedExceptionAndNotCallProductionExceptionHandler() {
+        final MockProducer<byte[], byte[]> mockProducer = new MockProducer<>(
+            cluster,
+            false,
+            new org.apache.kafka.clients.producer.RoundRobinPartitioner(),
+            new ByteArraySerializer(),
+            new ByteArraySerializer()
+        );
+        streamsProducer = new StreamsProducer(
+            mockProducer,
+            EXACTLY_ONCE_V2,
+            Time.SYSTEM,
+            logContext
+        );
+
+        final RecordCollector collector = newRecordCollector(new ProductionExceptionHandlerMock());
+        collector.initialize();
+
+        collector.send(topic, "key", "val", null, 0, null, stringSerializer, stringSerializer, sinkNodeName, context);
+        mockProducer.errorNext(new TransactionAbortedException()); // error out the send() call
+
+        collector.flush(); // need to call flush() to check for internal exceptions
+    }
+
     @Test
     public void shouldNotSendIfSendOfOtherTaskFailedInCallback() {
         final TaskId taskId1 = new TaskId(0, 0);