Skip to content

KAFKA-10813: InvalidProducerEpoch should be caught and throw as TaskMigrated #9700

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
abbccdda merged 1 commit into from
Dec 10, 2020
Merged

KAFKA-10813: InvalidProducerEpoch should be caught and throw as TaskMigrated #9700

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -605,6 +605,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
Expand Down Expand Up @@ -743,6 +745,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
Expand Down
5 changes: 4 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,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;
Expand Down Expand Up @@ -199,7 +200,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) {
Comment on lines +203 to +205

@ableegoldman ableegoldman Dec 7, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is that here, we catch these three exceptions -- ProducerFenced, InvalidProducerEpoch, and OutOfOrderSequence -- and wrap them as TaskMigratedException, while in StreamsProducer#send, we catch ProducerFenced, InvalidProducerEpoch, and UnknownProducerId exceptions and wrap those as TaskMigrated?

@ableegoldman ableegoldman Dec 7, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just to clarify, I think my question/confusion is twofold:

  1. Why is it OutOfOrderSequence in one place and UnknownProducerId in another?
  2. What is the difference between these two code paths? Is it really possible for example for the ProducerFencedException to sometimes be thrown directly from Producer#send, and sometimes be passed along through the callback ( as in streamsProducer.send(serializedRecord, (metadata, exception) )

@abbccdda abbccdda Dec 8, 2020

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wasn't over thinking here since this is a blocker-fix, so we don't want to trigger any regression here by changing existing exception catching logic. Would be good to do this as a follow-up I guess.

@ableegoldman ableegoldman Dec 8, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok SG, agree we should keep things simple since this is a last minute blocker

@guozhangwang guozhangwang Dec 9, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I thought about this a bit, and I think both UnknownProducerId and OutOfOrderSequence could be possibly thrown from the caller or directly (though in the later case they would be wrapped as KafkaException).

I created two tickets, one for producer and one for streams to improve the general picture moving forward.

https://issues.apache.org/jira/browse/KAFKA-10829
https://issues.apache.org/jira/browse/KAFKA-10830

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));
Expand Down
11 changes: 6 additions & 5 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,12 +184,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) {
Comment thread
ableegoldman marked this conversation as resolved.
throw new TaskMigratedException(
formatException("Producer got fenced trying to begin a new transaction"),
error
Expand All @@ -212,7 +212,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()
Expand All @@ -228,6 +228,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;
}

Expand All @@ -236,7 +237,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"));
}
Expand Down Expand Up @@ -279,7 +280,7 @@ void abortTransaction() {
" Will rely on broker to eventually abort the transaction after the transaction timeout passed.",
logAndSwallow
);
} 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
Expand Down
54 changes: 29 additions & 25 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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
Expand Down
73 changes: 46 additions & 27 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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!");
Expand All @@ -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,
Expand Down Expand Up @@ -1013,12 +1023,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();
Expand Down