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 index 1ce0b9328e141..75b7e102325f8 100644 --- 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 index dffa3e6e7003d..5c1bc7dc42e77 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 @@ -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 index e0469f68550e6..8554d72c4e1ce 100644 --- 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 send(final ProducerRecord 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 send(final ProducerRecord 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 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 index 1e3d232298036..701fc1cc12b4c 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 @@ -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 index 0747885712418..1d520922ed486 100644 --- 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();