apache · abbccdda · Dec 10, 2020 · Dec 10, 2020
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -606,6 +606,8 @@ public void initTransactions() {
      * @throws IllegalStateException if no transactional.id has been configured or if {@link #initTransactions()}
      *         has not yet been invoked
      * @throws ProducerFencedException if another producer with the same transactional.id is active
+     * @throws org.apache.kafka.common.errors.InvalidProducerEpochException if the producer has attempted to produce with an old epoch
+     *         to the partition leader. See the exception for more details
      * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
      *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
@@ -744,6 +746,8 @@ public void commitTransaction() throws ProducerFencedException {
      *
      * @throws IllegalStateException if no transactional.id has been configured or no transaction has been started
      * @throws ProducerFencedException fatal error indicating another producer with the same transactional.id is active
+     * @throws org.apache.kafka.common.errors.InvalidProducerEpochException if the producer has attempted to produce with an old epoch
+     *         to the partition leader. See the exception for more details
      * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
      *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured

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
@@ -24,6 +24,7 @@
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.AuthorizationException;
+import org.apache.kafka.common.errors.InvalidProducerEpochException;
 import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.OffsetMetadataTooLarge;
 import org.apache.kafka.common.errors.OutOfOrderSequenceException;
@@ -197,7 +198,9 @@ private void recordSendError(final String topic, final Exception exception, fina
         if (isFatalException(exception)) {
             errorMessage += "\nWritten offsets would not be recorded and no more records would be sent since this is a fatal error.";
             sendException.set(new StreamsException(errorMessage, exception));
-        } else if (exception instanceof ProducerFencedException || exception instanceof OutOfOrderSequenceException) {
+        } else if (exception instanceof ProducerFencedException ||
+                exception instanceof InvalidProducerEpochException ||
+                exception instanceof OutOfOrderSequenceException) {
             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, exception));

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java
@@ -183,12 +183,12 @@ public void resetProducer() {
         transactionInitialized = false;
     }
 
-    private void maybeBeginTransaction() throws ProducerFencedException {
+    private void maybeBeginTransaction() {
         if (eosEnabled() && !transactionInFlight) {
             try {
                 producer.beginTransaction();
                 transactionInFlight = true;
-            } catch (final ProducerFencedException error) {
+            } catch (final ProducerFencedException | InvalidProducerEpochException error) {
                 throw new TaskMigratedException(
                     formatException("Producer got fenced trying to begin a new transaction"),
                     error
@@ -211,7 +211,7 @@ Future<RecordMetadata> send(final ProducerRecord<byte[], byte[]> record,
             if (isRecoverable(uncaughtException)) {
                 // producer.send() call may throw a KafkaException which wraps a FencedException,
                 // in this case we should throw its wrapped inner cause so that it can be
-                // captured and re-wrapped as TaskMigrationException
+                // captured and re-wrapped as TaskMigratedException
                 throw new TaskMigratedException(
                     formatException("Producer got fenced trying to send a record"),
                     uncaughtException.getCause()
@@ -227,6 +227,7 @@ Future<RecordMetadata> send(final ProducerRecord<byte[], byte[]> record,
 
     private static boolean isRecoverable(final KafkaException uncaughtException) {
         return uncaughtException.getCause() instanceof ProducerFencedException ||
+            uncaughtException.getCause() instanceof InvalidProducerEpochException ||
             uncaughtException.getCause() instanceof UnknownProducerIdException;
     }
 
@@ -235,7 +236,7 @@ private static boolean isRecoverable(final KafkaException uncaughtException) {
      * @throws TaskMigratedException
      */
     void commitTransaction(final Map<TopicPartition, OffsetAndMetadata> offsets,
-                           final ConsumerGroupMetadata consumerGroupMetadata) throws ProducerFencedException {
+                           final ConsumerGroupMetadata consumerGroupMetadata) {
         if (!eosEnabled()) {
             throw new IllegalStateException(formatException("Exactly-once is not enabled"));
         }
@@ -270,7 +271,7 @@ void abortTransaction() throws ProducerFencedException {
         if (transactionInFlight) {
             try {
                 producer.abortTransaction();
-            } catch (final ProducerFencedException error) {
+            } catch (final ProducerFencedException | InvalidProducerEpochException error) {
                 // The producer is aborting the txn when there's still an ongoing one,
                 // which means that we did not commit the task while closing it, which
                 // means that it is a dirty close. Therefore it is possible that the dirty

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
@@ -30,6 +30,7 @@
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.AuthenticationException;
+import org.apache.kafka.common.errors.InvalidProducerEpochException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
@@ -449,7 +450,15 @@ public void shouldThrowInformativeStreamsExceptionOnValueAndNullKeyClassCastExce
 
     @Test
     public void shouldThrowTaskMigratedExceptionOnSubsequentSendWhenProducerFencedInCallback() {
-        final KafkaException exception = new ProducerFencedException("KABOOM!");
+        testThrowTaskMigratedExceptionOnSubsequentSend(new ProducerFencedException("KABOOM!"));
+    }
+
+    @Test
+    public void shouldThrowTaskMigratedExceptionOnSubsequentSendWhenInvalidEpochInCallback() {
+        testThrowTaskMigratedExceptionOnSubsequentSend(new InvalidProducerEpochException("KABOOM!"));
+    }
+
+    private void testThrowTaskMigratedExceptionOnSubsequentSend(final RuntimeException exception) {
         final RecordCollector collector = new RecordCollectorImpl(
             logContext,
             taskId,
@@ -463,21 +472,22 @@ public void shouldThrowTaskMigratedExceptionOnSubsequentSendWhenProducerFencedIn
 
         final TaskMigratedException thrown = assertThrows(
             TaskMigratedException.class, () ->
-                collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner)
+                                             collector.send(topic, "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner)
         );
         assertEquals(exception, thrown.getCause());
-        assertThat(
-            thrown.getMessage(),
-            equalTo("Error encountered sending record to topic topic for task 0_0 due to:" +
-                        "\norg.apache.kafka.common.errors.ProducerFencedException: KABOOM!" +
-                        "\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; it means all tasks belonging to this thread should be migrated.")
-        );
     }
 
     @Test
     public void shouldThrowTaskMigratedExceptionOnSubsequentFlushWhenProducerFencedInCallback() {
-        final KafkaException exception = new ProducerFencedException("KABOOM!");
+        testThrowTaskMigratedExceptionOnSubsequentFlush(new ProducerFencedException("KABOOM!"));
+    }
+
+    @Test
+    public void shouldThrowTaskMigratedExceptionOnSubsequentFlushWhenInvalidEpochInCallback() {
+        testThrowTaskMigratedExceptionOnSubsequentFlush(new InvalidProducerEpochException("KABOOM!"));
+    }
+
+    private void testThrowTaskMigratedExceptionOnSubsequentFlush(final RuntimeException exception) {
         final RecordCollector collector = new RecordCollectorImpl(
             logContext,
             taskId,
@@ -491,18 +501,19 @@ public void shouldThrowTaskMigratedExceptionOnSubsequentFlushWhenProducerFencedI
 
         final TaskMigratedException thrown = assertThrows(TaskMigratedException.class, collector::flush);
         assertEquals(exception, thrown.getCause());
-        assertThat(
-            thrown.getMessage(),
-            equalTo("Error encountered sending record to topic topic for task 0_0 due to:" +
-                        "\norg.apache.kafka.common.errors.ProducerFencedException: KABOOM!" +
-                        "\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; it means all tasks belonging to this thread should be migrated.")
-        );
     }
 
     @Test
     public void shouldThrowTaskMigratedExceptionOnSubsequentCloseWhenProducerFencedInCallback() {
-        final KafkaException exception = new ProducerFencedException("KABOOM!");
+        testThrowTaskMigratedExceptionOnSubsequentClose(new ProducerFencedException("KABOOM!"));
+    }
+
+    @Test
+    public void shouldThrowTaskMigratedExceptionOnSubsequentCloseWhenInvalidEpochInCallback() {
+        testThrowTaskMigratedExceptionOnSubsequentClose(new InvalidProducerEpochException("KABOOM!"));
+    }
+
+    private void testThrowTaskMigratedExceptionOnSubsequentClose(final RuntimeException exception) {
         final RecordCollector collector = new RecordCollectorImpl(
             logContext,
             taskId,
@@ -516,13 +527,6 @@ public void shouldThrowTaskMigratedExceptionOnSubsequentCloseWhenProducerFencedI
 
         final TaskMigratedException thrown = assertThrows(TaskMigratedException.class, collector::closeClean);
         assertEquals(exception, thrown.getCause());
-        assertThat(
-            thrown.getMessage(),
-            equalTo("Error encountered sending record to topic topic for task 0_0 due to:" +
-                        "\norg.apache.kafka.common.errors.ProducerFencedException: KABOOM!" +
-                        "\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; it means all tasks belonging to this thread should be migrated.")
-        );
     }
 
     @Test

diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java
@@ -855,10 +855,18 @@ public void shouldFailOnEosBeginTxnFatal() {
     }
 
     @Test
-    public void shouldThrowTaskMigratedExceptionOnEosSendFenced() {
-        // cannot use `eosMockProducer.fenceProducer()` because this would already trigger in `beginTransaction()`
-        final ProducerFencedException exception = new ProducerFencedException("KABOOM!");
+    public void shouldThrowTaskMigratedExceptionOnEosSendProducerFenced() {
+        testThrowTaskMigratedExceptionOnEosSend(new ProducerFencedException("KABOOM!"));
+    }
+
+    @Test
+    public void shouldThrowTaskMigratedExceptionOnEosSendInvalidEpoch() {
+        testThrowTaskMigratedExceptionOnEosSend(new InvalidProducerEpochException("KABOOM!"));
+    }
+
+    private void testThrowTaskMigratedExceptionOnEosSend(final RuntimeException exception) {
         // we need to mimic that `send()` always wraps error in a KafkaException
+        // cannot use `eosMockProducer.fenceProducer()` because this would already trigger in `beginTransaction()`
         eosAlphaMockProducer.sendException = new KafkaException(exception);
 
         final TaskMigratedException thrown = assertThrows(
@@ -894,9 +902,20 @@ public void shouldThrowTaskMigratedExceptionOnEosSendUnknownPid() {
     }
 
     @Test
-    public void shouldThrowTaskMigrateExceptionOnEosSendOffsetFenced() {
+    public void shouldThrowTaskMigrateExceptionOnEosSendOffsetProducerFenced() {
+        // cannot use `eosMockProducer.fenceProducer()` because this would already trigger in `beginTransaction()`
+        testThrowTaskMigrateExceptionOnEosSendOffset(new ProducerFencedException("KABOOM!"));
+    }
+
+    @Test
+    public void shouldThrowTaskMigrateExceptionOnEosSendOffsetInvalidEpoch() {
         // cannot use `eosMockProducer.fenceProducer()` because this would already trigger in `beginTransaction()`
-        eosAlphaMockProducer.sendOffsetsToTransactionException = new ProducerFencedException("KABOOM!");
+        testThrowTaskMigrateExceptionOnEosSendOffset(new InvalidProducerEpochException("KABOOM!"));
+    }
+
+    private void testThrowTaskMigrateExceptionOnEosSendOffset(final RuntimeException exception) {
+        // cannot use `eosMockProducer.fenceProducer()` because this would already trigger in `beginTransaction()`
+        eosAlphaMockProducer.sendOffsetsToTransactionException = exception;
 
         final TaskMigratedException thrown = assertThrows(
             TaskMigratedException.class,
@@ -931,24 +950,6 @@ public void shouldThrowStreamsExceptionOnEosSendOffsetError() {
         );
     }
 
-    @Test
-    public void shouldThrowTaskMigratedExceptionOnEosWithInvalidProducerEpoch() {
-        eosAlphaMockProducer.commitTransactionException = new InvalidProducerEpochException("KABOOM!");
-
-        final TaskMigratedException thrown = assertThrows(
-            TaskMigratedException.class,
-            () -> eosAlphaStreamsProducer.commitTransaction(offsetsAndMetadata, new ConsumerGroupMetadata("appId"))
-        );
-
-        assertThat(eosAlphaMockProducer.sentOffsets(), is(true));
-        assertThat(thrown.getCause(), is(eosAlphaMockProducer.commitTransactionException));
-        assertThat(
-            thrown.getMessage(),
-            is("Producer got fenced trying to commit a transaction [test];" +
-                   " it means all tasks belonging to this thread should be migrated.")
-        );
-    }
-
     @Test
     public void shouldFailOnEosSendOffsetFatal() {
         eosAlphaMockProducer.sendOffsetsToTransactionException = new RuntimeException("KABOOM!");
@@ -964,9 +965,18 @@ public void shouldFailOnEosSendOffsetFatal() {
     }
 
     @Test
-    public void shouldThrowTaskMigrateExceptionOnEosCommitTxFenced() {
+    public void shouldThrowTaskMigratedExceptionOnEosCommitWithProducerFenced() {
+        testThrowTaskMigratedExceptionOnEos(new ProducerFencedException("KABOOM!"));
+    }
+
+    @Test
+    public void shouldThrowTaskMigratedExceptionOnEosCommitWithInvalidEpoch() {
+        testThrowTaskMigratedExceptionOnEos(new InvalidProducerEpochException("KABOOM!"));
+    }
+
+    private void testThrowTaskMigratedExceptionOnEos(final RuntimeException exception) {
         // cannot use `eosMockProducer.fenceProducer()` because this would already trigger in `beginTransaction()`
-        eosAlphaMockProducer.commitTransactionException = new ProducerFencedException("KABOOM!");
+        eosAlphaMockProducer.commitTransactionException = exception;
 
         final TaskMigratedException thrown = assertThrows(
             TaskMigratedException.class,
@@ -1028,12 +1038,21 @@ public void shouldFailOnEosCommitTxFatal() {
     }
 
     @Test
-    public void shouldSwallowExceptionOnEosAbortTxFenced() {
+    public void shouldSwallowExceptionOnEosAbortTxProducerFenced() {
+        testSwallowExceptionOnEosAbortTx(new ProducerFencedException("KABOOM!"));
+    }
+
+    @Test
+    public void shouldSwallowExceptionOnEosAbortTxInvalidEpoch() {
+        testSwallowExceptionOnEosAbortTx(new InvalidProducerEpochException("KABOOM!"));
+    }
+
+    private void testSwallowExceptionOnEosAbortTx(final RuntimeException exception) {
         mockedProducer.initTransactions();
         mockedProducer.beginTransaction();
         expect(mockedProducer.send(record, null)).andReturn(null);
         mockedProducer.abortTransaction();
-        expectLastCall().andThrow(new ProducerFencedException("KABOOM!"));
+        expectLastCall().andThrow(exception);
         replay(mockedProducer);
 
         eosAlphaStreamsProducerWithMock.initTransaction();