diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java index ede47ba39c806..a18b2b6a3590a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java @@ -23,6 +23,7 @@ import org.apache.kafka.streams.errors.LockException; import org.apache.kafka.streams.errors.ProcessorStateException; import org.apache.kafka.streams.errors.StreamsException; +import org.apache.kafka.streams.errors.TaskMigratedException; import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.processor.TaskId; @@ -172,7 +173,13 @@ public String toString(final String indent) { * Flush all state stores owned by this task */ void flushState() { - stateMgr.flush(); + try { + stateMgr.flush(); + } catch (final ProcessorStateException e) { + if (e.getCause() instanceof RecoverableClientException) { + throw new TaskMigratedException(this, e); + } + } } /** diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasks.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasks.java index 9440c4c12015a..cca6c3dc921f1 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasks.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasks.java @@ -148,7 +148,7 @@ private RuntimeException suspendRunningTasks(final Set runningTasksToSus // swallow and move on since we are rebalancing log.info("Failed to suspend stream task {} since it got migrated to another thread already. " + "Closing it as zombie and moving on.", id); - firstException.compareAndSet(null, closeZombieTask(task)); + tryCloseZombieTask(task); prevActiveTasks.remove(id); } catch (final RuntimeException e) { log.error("Suspending stream task {} failed due to the following error:", id, e); diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java index 45ac26b1f65ec..125d3e315e456 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java @@ -97,14 +97,12 @@ Collection running() { return running.values(); } - RuntimeException closeZombieTask(final T task) { + void tryCloseZombieTask(final T task) { try { task.close(false, true); } catch (final RuntimeException e) { log.warn("Failed to close zombie {} {} due to {}; ignore and proceed.", taskTypeName, task.id(), e.toString()); - return e; } - return null; } boolean hasRunningTasks() { @@ -259,7 +257,7 @@ void shutdown(final boolean clean) { } catch (final TaskMigratedException e) { log.info("Failed to close {} {} since it got migrated to another thread already. " + "Closing it as zombie and move on.", taskTypeName, task.id()); - firstException.compareAndSet(null, closeZombieTask(task)); + tryCloseZombieTask(task); } catch (final RuntimeException t) { log.error("Failed while closing {} {} due to the following error:", task.getClass().getSimpleName(), diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java index 4d27e82ba2a59..9a32ce003947e 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java @@ -101,6 +101,9 @@ public void update(final ConsumerRecord record) { } public void flushState() { + // this could theoretically throw a ProcessorStateException caused by a ProducerFencedException, + // but in practice this shouldn't happen for global state update tasks, since the stores are not + // logged and there are no downstream operators after global stores. stateMgr.flush(); stateMgr.checkpoint(offsets); } 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 8430ff1be917b..d72115e52a4b2 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.SecurityDisabledException; import org.apache.kafka.common.errors.SerializationException; import org.apache.kafka.common.errors.TimeoutException; +import org.apache.kafka.common.errors.UnknownProducerIdException; import org.apache.kafka.common.errors.UnknownServerException; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.header.Headers; @@ -179,13 +180,13 @@ public void onCompletion(final RecordMetadata metadata, offsets.put(tp, metadata.offset()); } else { if (sendException == null) { - if (exception instanceof ProducerFencedException) { + if (exception instanceof ProducerFencedException || exception instanceof UnknownProducerIdException) { log.warn(LOG_MESSAGE, topic, exception.getMessage(), exception); // KAFKA-7510 put message key and value in TRACE level log so we don't leak data by default log.trace("Failed message: (key {} value {} timestamp {}) topic=[{}] partition=[{}]", key, value, timestamp, topic, partition); - sendException = new ProducerFencedException( + sendException = new RecoverableClientException( String.format( EXCEPTION_MESSAGE, logPrefix, @@ -193,7 +194,8 @@ public void onCompletion(final RecordMetadata metadata, timestamp, topic, exception.toString() - ) + ), + exception ); } else { if (productionExceptionIsFatal(exception)) { @@ -233,13 +235,13 @@ public void onCompletion(final RecordMetadata metadata, String.format("%sFailed to send record to topic %s due to timeout.", logPrefix, topic), e ); - } catch (final Exception uncaughtException) { + } catch (final RuntimeException uncaughtException) { if (uncaughtException instanceof KafkaException && uncaughtException.getCause() instanceof ProducerFencedException) { final KafkaException kafkaException = (KafkaException) 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 - throw (ProducerFencedException) kafkaException.getCause(); + throw new RecoverableClientException("Caught a wrapped ProducerFencedException", kafkaException); } else { throw new StreamsException( String.format( @@ -264,7 +266,11 @@ private void checkForException() { @Override public void flush() { log.debug("Flushing producer"); - producer.flush(); + try { + producer.flush(); + } catch (final ProducerFencedException | UnknownProducerIdException e) { + throw new RecoverableClientException("Caught a recoverable exception while flushing", e); + } checkForException(); } @@ -272,7 +278,11 @@ public void flush() { public void close() { log.debug("Closing producer"); if (producer != null) { - producer.close(); + try { + producer.close(); + } catch (final ProducerFencedException | UnknownProducerIdException e) { + throw new RecoverableClientException("Caught a recoverable exception while closing", e); + } producer = null; } checkForException(); diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecoverableClientException.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecoverableClientException.java new file mode 100644 index 0000000000000..c4a96f7dcc622 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecoverableClientException.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.processor.internals; + +import org.apache.kafka.common.KafkaException; + +/** + * Denotes an exception that is recoverable by re-creating the client (ie, the client is no longer in a valid state), + * as opposed to retriable (the failure was transient, so the same client can be used again later), + * or fatal (the request was actually invalid, so retrying or recovering would not help) + * + * This class also serves the dual purpose of capturing the stack trace as early as possible, + * at the site of the Producer call, since the exeptions that cause this don't record stack traces. + */ +public class RecoverableClientException extends KafkaException { + public RecoverableClientException(final String message, final Throwable cause) { + super(message, cause); + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java index 251d99fad2b37..760c4ae68d9f2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java @@ -122,6 +122,9 @@ public void commit() { } private void flushAndCheckpointState() { + // this could theoretically throw a ProcessorStateException caused by a ProducerFencedException, + // but in practice this shouldn't happen for standby tasks, since they don't produce to changelog topics + // or downstream topics. stateMgr.flush(); stateMgr.checkpoint(Collections.emptyMap()); } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java index fcab9f1094494..950d4a5e87c82 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java @@ -27,6 +27,7 @@ import org.apache.kafka.common.errors.AuthorizationException; import org.apache.kafka.common.errors.ProducerFencedException; import org.apache.kafka.common.errors.TimeoutException; +import org.apache.kafka.common.errors.UnknownProducerIdException; import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Avg; @@ -337,8 +338,8 @@ public void initializeTopology() { if (eosEnabled) { try { this.producer.beginTransaction(); - } catch (final ProducerFencedException fatal) { - throw new TaskMigratedException(this, fatal); + } catch (final ProducerFencedException | UnknownProducerIdException e) { + throw new TaskMigratedException(this, e); } transactionInFlight = true; } @@ -438,8 +439,8 @@ public boolean process() { if (recordInfo.queue().size() == maxBufferedSize) { consumer.resume(singleton(partition)); } - } catch (final ProducerFencedException fatal) { - throw new TaskMigratedException(this, fatal); + } catch (final RecoverableClientException e) { + throw new TaskMigratedException(this, e); } catch (final KafkaException e) { final String stackTrace = getStacktraceString(e); throw new StreamsException(format("Exception caught in process. taskId=%s, " + @@ -488,8 +489,8 @@ public void punctuate(final ProcessorNode node, final long timestamp, final Punc try { node.punctuate(timestamp, punctuator); - } catch (final ProducerFencedException fatal) { - throw new TaskMigratedException(this, fatal); + } catch (final RecoverableClientException e) { + throw new TaskMigratedException(this, e); } catch (final KafkaException e) { throw new StreamsException(String.format("%sException caught while punctuating processor '%s'", logPrefix, node.name()), e); } finally { @@ -558,7 +559,7 @@ void commit(final boolean startNewTransaction, final Map p } else { consumer.commitSync(consumedOffsetsAndMetadata); } - } catch (final CommitFailedException | ProducerFencedException error) { + } catch (final CommitFailedException | ProducerFencedException | UnknownProducerIdException error) { throw new TaskMigratedException(this, error); } @@ -581,8 +582,8 @@ protected void flushState() { super.flushState(); try { recordCollector.flush(); - } catch (final ProducerFencedException fatal) { - throw new TaskMigratedException(this, fatal); + } catch (final RecoverableClientException e) { + throw new TaskMigratedException(this, e); } } @@ -664,7 +665,7 @@ void suspend(final boolean clean, try { recordCollector.close(); - } catch (final ProducerFencedException e) { + } catch (final RecoverableClientException e) { taskMigratedException = new TaskMigratedException(this, e); } finally { producer = null; diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java index d04b7c25eb543..96ffe3beb157e 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java @@ -188,10 +188,11 @@ void openDB(final ProcessorContext context) { throw new ProcessorStateException(fatal); } - maybeSetUpMetricsRecorder(context, configs); - openRocksDB(dbOptions, columnFamilyOptions); open = true; + + // Do this last because the prior operations could throw exceptions. + maybeSetUpMetricsRecorder(context, configs); } private void maybeSetUpMetricsRecorder(final ProcessorContext context, final Map configs) { diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java index 6d98ada278ca3..35db23d8446ea 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java @@ -37,6 +37,7 @@ import org.easymock.EasyMock; import org.junit.Before; import org.junit.Test; +import org.junit.function.ThrowingRunnable; import java.util.ArrayList; import java.util.Collection; @@ -44,8 +45,8 @@ import java.util.HashSet; import java.util.List; import java.util.Set; -import org.junit.function.ThrowingRunnable; +import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.CoreMatchers.not; import static org.hamcrest.CoreMatchers.nullValue; import static org.hamcrest.MatcherAssert.assertThat; @@ -235,7 +236,7 @@ public void shouldCloseTaskOnSuspendIfTaskMigratedException() { t1.suspend(); EasyMock.expectLastCall().andThrow(new TaskMigratedException()); t1.close(false, true); - EasyMock.expectLastCall(); + EasyMock.expectLastCall().andThrow(new RuntimeException("any exception")); EasyMock.replay(t1); assertThat(suspendTask(), nullValue()); @@ -243,6 +244,37 @@ public void shouldCloseTaskOnSuspendIfTaskMigratedException() { EasyMock.verify(t1); } + @Test + public void shouldCloseUncleanAndThenRethrowOnShutdownIfRuntimeException() { + mockTaskInitialization(); + + t1.close(true, false); + EasyMock.expectLastCall().andThrow(new RuntimeException("any first exception")); + t1.close(false, false); + EasyMock.expectLastCall().andThrow(new RuntimeException("any second exception")); + EasyMock.replay(t1); + addAndInitTask(); + try { + assignedTasks.shutdown(true); + fail("expected a runtime exception"); + } catch (final RuntimeException e) { + assertThat(e.getMessage(), is("any first exception")); + } + } + + @Test + public void shouldCloseWithoutExceptionOnShutdownIfTaskMigratedException() { + mockTaskInitialization(); + + t1.close(true, false); + EasyMock.expectLastCall().andThrow(new TaskMigratedException()); + t1.close(false, true); + EasyMock.expectLastCall().andThrow(new RuntimeException("any second exception")); + EasyMock.replay(t1); + addAndInitTask(); + assignedTasks.shutdown(true); + } + @Test public void shouldResumeMatchingSuspendedTasks() { mockRunningTaskSuspension(); @@ -603,7 +635,7 @@ public void action(final StreamTask task) { public Set taskIds() { return assignedTasks.suspendedTaskIds(); } - + }.createTaskAndClear(); } 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 f40f3c1871c41..72d315f658376 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 @@ -27,6 +27,7 @@ import org.apache.kafka.common.Node; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.ProducerFencedException; import org.apache.kafka.common.header.Header; import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeader; @@ -191,6 +192,26 @@ public synchronized Future send(final ProducerRecord record, fin collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); } + @SuppressWarnings("unchecked") + @Test(expected = RecoverableClientException.class) + public void shouldThrowRecoverableExceptionWhenProducerFencedInCallback() { + final RecordCollector collector = new RecordCollectorImpl( + "test", + logContext, + new DefaultProductionExceptionHandler(), + new Metrics().sensor("skipped-records")); + collector.init(new MockProducer(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer) { + @Override + public synchronized Future send(final ProducerRecord record, final Callback callback) { + callback.onCompletion(null, new ProducerFencedException("asdf")); + return null; + } + }); + + collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); + } + @SuppressWarnings("unchecked") @Test public void shouldThrowStreamsExceptionOnSubsequentCallIfASendFailsWithDefaultExceptionHandler() { diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java index 59cc2a52beb07..e1252d3c48c78 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java @@ -22,8 +22,11 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.clients.producer.MockProducer; +import org.apache.kafka.clients.producer.internals.DefaultPartitioner; +import org.apache.kafka.common.Cluster; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.MetricName; +import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.AuthorizationException; import org.apache.kafka.common.errors.ProducerFencedException; @@ -51,7 +54,9 @@ import org.apache.kafka.streams.processor.Punctuator; import org.apache.kafka.streams.processor.StateRestoreListener; import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.processor.StreamPartitioner; import org.apache.kafka.streams.processor.TaskId; +import org.apache.kafka.streams.processor.WallclockTimestampExtractor; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender; import org.apache.kafka.streams.state.internals.OffsetCheckpoint; @@ -1090,7 +1095,7 @@ public void shouldOnlyCloseProducerIfFencedOnCommitDuringCleanCloseWithEosEnable fail("should have thrown TaskMigratedException"); } catch (final TaskMigratedException expected) { task = null; - assertTrue(expected.getCause() instanceof ProducerFencedException); + assertTrue(expected.getCause() instanceof RecoverableClientException); } assertFalse(producer.transactionCommitted()); @@ -1111,7 +1116,7 @@ public void shouldNotCloseProducerIfFencedOnCloseDuringCleanCloseWithEosEnabled( fail("should have thrown TaskMigratedException"); } catch (final TaskMigratedException expected) { task = null; - assertTrue(expected.getCause() instanceof ProducerFencedException); + assertTrue(expected.getCause() instanceof RecoverableClientException); } assertTrue(producer.transactionCommitted()); @@ -1158,6 +1163,190 @@ public void shouldOnlyCloseProducerIfFencedOnAbortDuringUncleanCloseWithEosEnabl assertTrue(producer.closed()); } + @Test + public void shouldMigrateTaskIfFencedDuringFlush() { + final StateStore stateStore = new MockKeyValueStore(storeName, true, true); + + final Map storeToChangelogTopic = true ? Collections.singletonMap(storeName, storeName + "-changelog") : Collections.emptyMap(); + final ProcessorTopology topology1 = new ProcessorTopology( + asList(source1, source2), + mkMap(mkEntry(topic1, source1), mkEntry(topic2, source2)), + Collections.emptyMap(), + singletonList(stateStore), + Collections.emptyList(), + storeToChangelogTopic, + Collections.emptySet() + ); + + task = new StreamTask( + taskId00, + partitions, + topology1, + consumer, + changelogReader, + createConfig(true), + streamsMetrics, + stateDirectory, + null, + time, + () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer) + ); + task.initializeStateStores(); + task.initializeTopology(); + producer.fenceProducer(); + + try { + task.flushState(); + fail("Expected a TaskMigratedException"); + } catch (final TaskMigratedException expected) { + assertThat(expected.migratedTask(), is(task)); + } + } + + @Test + public void shouldMigrateTaskIfFencedDuringProcess() { + final StateStore stateStore = new MockKeyValueStore(storeName, true, true); + + final Map storeToChangelogTopic = true ? Collections.singletonMap(storeName, storeName + "-changelog") : Collections.emptyMap(); + final SourceNode sourceNode = new SourceNode<>("test", singletonList(topic1), new WallclockTimestampExtractor(), intDeserializer, intDeserializer); + final SinkNode sinkNode = new SinkNode<>("test-sink", new StaticTopicNameExtractor<>("out-topic"), intSerializer, intSerializer, new StreamPartitioner() { + @Override + public Integer partition(final String topic, + final Integer key, + final Integer value, + final int numPartitions) { + return 1; + } + }); + + sourceNode.addChild(sinkNode); + + final ProcessorTopology topology1 = new ProcessorTopology( + asList(sourceNode, sinkNode), + mkMap(mkEntry(topic1, sourceNode), mkEntry(topic2, sourceNode)), + mkMap(mkEntry("out-topic", sinkNode)), + singletonList(stateStore), + Collections.emptyList(), + storeToChangelogTopic, + Collections.emptySet() + ); + + task = new StreamTask( + taskId00, + partitions, + topology1, + consumer, + changelogReader, + createConfig(true), + streamsMetrics, + stateDirectory, + null, + time, + () -> producer = new MockProducer( + Cluster.empty(), + false, + new DefaultPartitioner(), + bytesSerializer, + bytesSerializer + ) { + @Override + public List partitionsFor(final String topic) { + return singletonList(null); + } + } + ); + task.initializeStateStores(); + task.initializeTopology(); + producer.fenceProducer(); + + try { + + task.addRecords(partition1, singletonList(getConsumerRecord(partition1, 0))); + task.process(); + fail("Expected a TaskMigratedException"); + } catch (final TaskMigratedException expected) { + assertThat(expected.migratedTask(), is(task)); + } + } + + @Test + public void shouldMigrateTaskIfFencedDuringPunctuate() { + task = createStatelessTask(createConfig(true)); + + final RecordCollectorImpl recordCollector = new RecordCollectorImpl("StreamTask", + new LogContext("StreamTaskTest "), new DefaultProductionExceptionHandler(), new Metrics().sensor("skipped-records")); + recordCollector.init(producer); + + task.initializeStateStores(); + task.initializeTopology(); + producer.fenceProducer(); + try { + task.punctuate( + processorSystemTime, + 5, + PunctuationType.WALL_CLOCK_TIME, + timestamp -> recordCollector.send( + "result-topic1", + 3, + 5, + null, + 0, + time.milliseconds(), + new IntegerSerializer(), + new IntegerSerializer() + ) + ); + fail("Expected a TaskMigratedException"); + } catch (final TaskMigratedException expected) { + assertThat(expected.migratedTask(), is(task)); + } + } + + @Test + public void shouldMigrateTaskIfFencedDuringCommit() { + final ProcessorTopology topology1 = withSources( + asList(source1, source2, processorStreamTime, processorSystemTime), + mkMap(mkEntry(topic1, source1), mkEntry(topic2, source2)) + ); + + source1.addChild(processorStreamTime); + source2.addChild(processorStreamTime); + source1.addChild(processorSystemTime); + source2.addChild(processorSystemTime); + + task = new StreamTask( + taskId00, + partitions, + topology1, + consumer, + changelogReader, + createConfig(true), + streamsMetrics, + stateDirectory, + null, + time, + () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer)) { + @Override + protected void flushState() { + // do nothing so that we actually make it to the commit interaction. + } + }; + + final RecordCollectorImpl recordCollector = new RecordCollectorImpl("StreamTask", + new LogContext("StreamTaskTest "), new DefaultProductionExceptionHandler(), new Metrics().sensor("skipped-records")); + recordCollector.init(producer); + + task.initializeStateStores(); + task.initializeTopology(); + producer.fenceProducer(); + try { + task.commit(); + fail("Expected a TaskMigratedException"); + } catch (final TaskMigratedException expected) { + assertThat(expected.migratedTask(), is(task)); + } + } + @Test public void shouldOnlyCloseFencedProducerOnUncleanClosedWithEosEnabled() { task = createStatelessTask(createConfig(true)); @@ -1283,7 +1472,7 @@ public void shouldWrapProducerFencedExceptionWithTaskMigragedExceptionInSuspendW task.suspend(); fail("Should have throws TaskMigratedException"); } catch (final TaskMigratedException expected) { - assertTrue(expected.getCause() instanceof ProducerFencedException); + assertTrue(expected.getCause() instanceof RecoverableClientException); } task = null; @@ -1300,7 +1489,7 @@ public void shouldWrapProducerFencedExceptionWithTaskMigragedExceptionInSuspendW task.suspend(); fail("Should have throws TaskMigratedException"); } catch (final TaskMigratedException expected) { - assertTrue(expected.getCause() instanceof ProducerFencedException); + assertTrue(expected.getCause() instanceof RecoverableClientException); } assertTrue(producer.transactionCommitted()); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskSuite.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskSuite.java new file mode 100644 index 0000000000000..564719088b634 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskSuite.java @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.processor.internals; + +import org.junit.runner.RunWith; +import org.junit.runners.Suite; + +/** + * This suite runs all the tests related to task management. It's intended to simplify feature testing from IDEs. + * + * If desired, it can also be added to a Gradle build task, although this isn't strictly necessary, since all + * these tests are already included in the `:streams:test` task. + */ +@RunWith(Suite.class) +@Suite.SuiteClasses({ + AbstractTaskTest.class, + StreamTaskTest.class, + StandbyTaskTest.class, + AssignedStreamsTasksTest.class, +}) +public class TaskSuite { +} + + diff --git a/streams/src/test/java/org/apache/kafka/test/MockKeyValueStore.java b/streams/src/test/java/org/apache/kafka/test/MockKeyValueStore.java index 11b73d92a2c8a..9eb475c5890c1 100644 --- a/streams/src/test/java/org/apache/kafka/test/MockKeyValueStore.java +++ b/streams/src/test/java/org/apache/kafka/test/MockKeyValueStore.java @@ -18,9 +18,11 @@ import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.IntegerDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.streams.processor.StateRestoreCallback; import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.processor.internals.RecordCollector; import org.apache.kafka.streams.state.KeyValueIterator; import org.apache.kafka.streams.state.KeyValueStore; @@ -41,11 +43,22 @@ public class MockKeyValueStore implements KeyValueStore { public boolean closed = true; public final ArrayList keys = new ArrayList<>(); public final ArrayList values = new ArrayList<>(); + private final boolean simulateForwardOnFlush; + private RecordCollector collector; public MockKeyValueStore(final String name, final boolean persistent) { this.name = name; this.persistent = persistent; + simulateForwardOnFlush = false; + } + + public MockKeyValueStore(final String name, + final boolean persistent, + final boolean simulateForwardOnFlush) { + this.name = name; + this.persistent = persistent; + this.simulateForwardOnFlush = simulateForwardOnFlush; } @Override @@ -57,12 +70,18 @@ public String name() { public void init(final ProcessorContext context, final StateStore root) { context.register(root, stateRestoreCallback); + if (simulateForwardOnFlush) { + collector = ((RecordCollector.Supplier) context).recordCollector(); + } initialized = true; closed = false; } @Override public void flush() { + if (simulateForwardOnFlush) { + collector.send("any", "anykey", "anyvalue", null, 0, 0L, new StringSerializer(), new StringSerializer()); + } instanceLastFlushCount.set(GLOBAL_FLUSH_COUNTER.getAndIncrement()); flushed = true; }