diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateMaintainer.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateMaintainer.java index acb32f7b6c9b2..9a8aab6eb3c84 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateMaintainer.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateMaintainer.java @@ -31,7 +31,7 @@ interface GlobalStateMaintainer { void flushState(); - void close() throws IOException; + void close(final boolean wipeStateStore) throws IOException; void update(ConsumerRecord record); } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java index 2f71eff2bb733..942308ba20e27 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java @@ -19,7 +19,6 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.consumer.InvalidOffsetException; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.TimeoutException; @@ -291,27 +290,17 @@ private void restoreState(final StateRestoreCallback stateRestoreCallback, long restoreCount = 0L; while (offset < highWatermark) { - try { - final ConsumerRecords records = globalConsumer.poll(pollTime); - final List> restoreRecords = new ArrayList<>(); - for (final ConsumerRecord record : records.records(topicPartition)) { - if (record.key() != null) { - restoreRecords.add(recordConverter.convert(record)); - } + final ConsumerRecords records = globalConsumer.poll(pollTime); + final List> restoreRecords = new ArrayList<>(); + for (final ConsumerRecord record : records.records(topicPartition)) { + if (record.key() != null) { + restoreRecords.add(recordConverter.convert(record)); } - offset = globalConsumer.position(topicPartition); - stateRestoreAdapter.restoreBatch(restoreRecords); - stateRestoreListener.onBatchRestored(topicPartition, storeName, offset, restoreRecords.size()); - restoreCount += restoreRecords.size(); - } catch (final InvalidOffsetException recoverableException) { - log.warn("Restoring GlobalStore {} failed due to: {}. Deleting global store to recreate from scratch.", - storeName, - recoverableException.toString()); - - // TODO K9113: we remove the re-init logic and push it to be handled by the thread directly - - restoreCount = 0L; } + offset = globalConsumer.position(topicPartition); + stateRestoreAdapter.restoreBatch(restoreRecords); + stateRestoreListener.onBatchRestored(topicPartition, storeName, offset, restoreRecords.size()); + restoreCount += restoreRecords.size(); } stateRestoreListener.onRestoreEnd(topicPartition, storeName, restoreCount); checkpointFileCache.put(topicPartition, offset); 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 ddef7a42f110a..b557330eb1075 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 @@ -19,8 +19,10 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Utils; import org.apache.kafka.streams.errors.DeserializationExceptionHandler; import org.apache.kafka.streams.errors.StreamsException; +import org.slf4j.Logger; import java.io.IOException; import java.util.HashMap; @@ -33,6 +35,8 @@ * Updates the state for all Global State Stores. */ public class GlobalStateUpdateTask implements GlobalStateMaintainer { + private final Logger log; + private final LogContext logContext; private final ProcessorTopology topology; private final InternalProcessorContext processorContext; @@ -40,18 +44,18 @@ public class GlobalStateUpdateTask implements GlobalStateMaintainer { private final Map deserializers = new HashMap<>(); private final GlobalStateManager stateMgr; private final DeserializationExceptionHandler deserializationExceptionHandler; - private final LogContext logContext; - public GlobalStateUpdateTask(final ProcessorTopology topology, + public GlobalStateUpdateTask(final LogContext logContext, + final ProcessorTopology topology, final InternalProcessorContext processorContext, final GlobalStateManager stateMgr, - final DeserializationExceptionHandler deserializationExceptionHandler, - final LogContext logContext) { + final DeserializationExceptionHandler deserializationExceptionHandler) { + this.logContext = logContext; + this.log = logContext.logger(getClass()); this.topology = topology; this.stateMgr = stateMgr; this.processorContext = processorContext; this.deserializationExceptionHandler = deserializationExceptionHandler; - this.logContext = logContext; } /** @@ -114,8 +118,16 @@ public void flushState() { stateMgr.checkpoint(offsets); } - public void close() throws IOException { + public void close(final boolean wipeStateStore) throws IOException { stateMgr.close(); + if (wipeStateStore) { + try { + log.info("Deleting global task directory after detecting corruption."); + Utils.delete(stateMgr.baseDir()); + } catch (final IOException e) { + log.error("Failed to delete global task directory after detecting corruption.", e); + } + } } private void initTopology() { diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java index 923480fa5b479..14d1ef89f660f 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java @@ -234,24 +234,18 @@ void initialize() { } void pollAndUpdate() { - try { - final ConsumerRecords received = globalConsumer.poll(pollTime); - for (final ConsumerRecord record : received) { - stateMaintainer.update(record); - } - final long now = time.milliseconds(); - if (now >= lastFlush + flushInterval) { - stateMaintainer.flushState(); - lastFlush = now; - } - } catch (final InvalidOffsetException recoverableException) { - log.error("Updating global state failed. You can restart KafkaStreams to recover from this error.", recoverableException); - throw new StreamsException("Updating global state failed. " + - "You can restart KafkaStreams to recover from this error.", recoverableException); + final ConsumerRecords received = globalConsumer.poll(pollTime); + for (final ConsumerRecord record : received) { + stateMaintainer.update(record); + } + final long now = time.milliseconds(); + if (now >= lastFlush + flushInterval) { + stateMaintainer.flushState(); + lastFlush = now; } } - public void close() throws IOException { + public void close(final boolean wipeStateStore) throws IOException { try { globalConsumer.close(); } catch (final RuntimeException e) { @@ -260,7 +254,7 @@ public void close() throws IOException { log.error("Failed to close global consumer due to the following error:", e); } - stateMaintainer.close(); + stateMaintainer.close(wipeStateStore); } } @@ -284,10 +278,21 @@ public void run() { } setState(State.RUNNING); + boolean wipeStateStore = false; try { while (stillRunning()) { stateConsumer.pollAndUpdate(); } + } catch (final InvalidOffsetException recoverableException) { + wipeStateStore = true; + log.error( + "Updating global state failed due to inconsistent local state. Will attempt to clean up the local state. You can restart KafkaStreams to recover from this error.", + recoverableException + ); + throw new StreamsException( + "Updating global state failed. You can restart KafkaStreams to recover from this error.", + recoverableException + ); } finally { // set the state to pending shutdown first as it may be called due to error; // its state may already be PENDING_SHUTDOWN so it will return false but we @@ -297,7 +302,7 @@ public void run() { log.info("Shutting down"); try { - stateConsumer.close(); + stateConsumer.close(wipeStateStore); } catch (final IOException e) { log.error("Failed to close state maintainer due to the following error:", e); } @@ -331,17 +336,36 @@ private StateConsumer initialize() { logContext, globalConsumer, new GlobalStateUpdateTask( + logContext, topology, globalProcessorContext, stateMgr, - config.defaultDeserializationExceptionHandler(), - logContext + config.defaultDeserializationExceptionHandler() ), time, Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG)), config.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG) ); - stateConsumer.initialize(); + + try { + stateConsumer.initialize(); + } catch (final InvalidOffsetException recoverableException) { + log.error( + "Bootstrapping global state failed due to inconsistent local state. Will attempt to clean up the local state. You can restart KafkaStreams to recover from this error.", + recoverableException + ); + + try { + stateConsumer.close(true); + } catch (final IOException e) { + log.error("Failed to close state consumer due to the following error:", e); + } + + throw new StreamsException( + "Bootstrapping global state failed. You can restart KafkaStreams to recover from this error.", + recoverableException + ); + } return stateConsumer; } catch (final LockException fatalException) { diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java index d8229b5f37d0a..d9374732473bb 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java @@ -17,7 +17,6 @@ package org.apache.kafka.streams.processor.internals; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.InvalidOffsetException; import org.apache.kafka.clients.consumer.MockConsumer; import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.common.PartitionInfo; @@ -323,21 +322,6 @@ public void shouldRestoreRecordsUpToHighwatermark() { assertEquals(2, stateRestoreCallback.restored.size()); } - @Test - public void shouldRecoverFromInvalidOffsetExceptionAndRestoreRecords() { - initializeConsumer(2, 0, t1); - consumer.setPollException(new InvalidOffsetException("Try Again!") { - public Set partitions() { - return Collections.singleton(t1); - } - }); - - stateManager.initialize(); - - stateManager.registerStore(store1, stateRestoreCallback); - assertEquals(2, stateRestoreCallback.restored.size()); - } - @Test public void shouldListenForRestoreEvents() { initializeConsumer(5, 1, t1); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java index 23191993a2ea3..42d3fc92530db 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java @@ -32,9 +32,11 @@ import org.apache.kafka.test.MockProcessorNode; import org.apache.kafka.test.MockSourceNode; import org.apache.kafka.test.NoOpProcessorContext; +import org.apache.kafka.test.TestUtils; import org.junit.Before; import org.junit.Test; +import java.io.File; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -44,6 +46,7 @@ import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -65,6 +68,7 @@ public class GlobalStateTaskTest { private final MockProcessorNode processorTwo = new MockProcessorNode<>(); private final Map offsets = new HashMap<>(); + private File testDirectory = TestUtils.tempDirectory("global-store"); private final NoOpProcessorContext context = new NoOpProcessorContext(); private ProcessorTopology topology; @@ -88,8 +92,14 @@ public void before() { offsets.put(t1, 50L); offsets.put(t2, 100L); - stateMgr = new GlobalStateManagerStub(storeNames, offsets); - globalStateTask = new GlobalStateUpdateTask(topology, context, stateMgr, new LogAndFailExceptionHandler(), logContext); + stateMgr = new GlobalStateManagerStub(storeNames, offsets, testDirectory); + globalStateTask = new GlobalStateUpdateTask( + logContext, + topology, + context, + stateMgr, + new LogAndFailExceptionHandler() + ); } @Test @@ -171,11 +181,11 @@ public void shouldThrowStreamsExceptionWhenValueDeserializationFails() { @Test public void shouldNotThrowStreamsExceptionWhenKeyDeserializationFailsWithSkipHandler() { final GlobalStateUpdateTask globalStateTask2 = new GlobalStateUpdateTask( + logContext, topology, context, stateMgr, - new LogAndContinueExceptionHandler(), - logContext + new LogAndContinueExceptionHandler() ); final byte[] key = new LongSerializer().serialize(topic2, 1L); final byte[] recordValue = new IntegerSerializer().serialize(topic2, 10); @@ -186,11 +196,11 @@ public void shouldNotThrowStreamsExceptionWhenKeyDeserializationFailsWithSkipHan @Test public void shouldNotThrowStreamsExceptionWhenValueDeserializationFails() { final GlobalStateUpdateTask globalStateTask2 = new GlobalStateUpdateTask( + logContext, topology, context, stateMgr, - new LogAndContinueExceptionHandler(), - logContext + new LogAndContinueExceptionHandler() ); final byte[] key = new IntegerSerializer().serialize(topic2, 1); final byte[] recordValue = new LongSerializer().serialize(topic2, 10L); @@ -221,4 +231,10 @@ public void shouldCheckpointOffsetsWhenStateIsFlushed() { assertThat(stateMgr.changelogOffsets(), equalTo(expectedOffsets)); } + @Test + public void shouldWipeGlobalStateDirectory() throws Exception { + assertTrue(stateMgr.baseDir().exists()); + globalStateTask.close(true); + assertFalse(stateMgr.baseDir().exists()); + } } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java index ddb20c951ca47..15bae376b704a 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java @@ -31,8 +31,8 @@ import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.internals.InternalNameProvider; import org.apache.kafka.streams.kstream.internals.KTableSource; -import org.apache.kafka.streams.kstream.internals.TimestampedKeyValueStoreMaterializer; import org.apache.kafka.streams.kstream.internals.MaterializedInternal; +import org.apache.kafka.streams.kstream.internals.TimestampedKeyValueStoreMaterializer; import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; import org.apache.kafka.streams.state.KeyValueStore; @@ -41,6 +41,7 @@ import org.junit.Before; import org.junit.Test; +import java.io.File; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -63,12 +64,12 @@ public class GlobalStreamThreadTest { private final MockStateRestoreListener stateRestoreListener = new MockStateRestoreListener(); private GlobalStreamThread globalStreamThread; private StreamsConfig config; + private String baseDirectoryName; private final static String GLOBAL_STORE_TOPIC_NAME = "foo"; private final static String GLOBAL_STORE_NAME = "bar"; private final TopicPartition topicPartition = new TopicPartition(GLOBAL_STORE_TOPIC_NAME, 0); - @SuppressWarnings("unchecked") @Before public void before() { final MaterializedInternal> materialized = @@ -97,10 +98,11 @@ public String newStoreName(final String prefix) { "processorName", new KTableSource<>(GLOBAL_STORE_NAME, GLOBAL_STORE_NAME)); + baseDirectoryName = TestUtils.tempDirectory().getAbsolutePath(); final HashMap properties = new HashMap<>(); properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "blah"); - properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "blah"); - properties.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath()); + properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "testAppId"); + properties.put(StreamsConfig.STATE_DIR_CONFIG, baseDirectoryName); config = new StreamsConfig(properties); globalStreamThread = new GlobalStreamThread( builder.rewriteTopology(config).buildGlobalStateTopology(), @@ -128,10 +130,9 @@ public void shouldThrowStreamsExceptionOnStartupIfThereIsAStreamsException() { assertFalse(globalStreamThread.stillRunning()); } - @SuppressWarnings("unchecked") @Test public void shouldThrowStreamsExceptionOnStartupIfExceptionOccurred() { - final MockConsumer mockConsumer = new MockConsumer(OffsetResetStrategy.EARLIEST) { + final MockConsumer mockConsumer = new MockConsumer(OffsetResetStrategy.EARLIEST) { @Override public List partitionsFor(final String topic) { throw new RuntimeException("KABOOM!"); @@ -186,7 +187,6 @@ public void shouldCloseStateStoresOnClose() throws Exception { assertFalse(globalStore.isOpen()); } - @SuppressWarnings("unchecked") @Test public void shouldTransitionToDeadOnClose() throws Exception { initializeConsumer(); @@ -197,7 +197,6 @@ public void shouldTransitionToDeadOnClose() throws Exception { assertEquals(GlobalStreamThread.State.DEAD, globalStreamThread.state()); } - @SuppressWarnings("unchecked") @Test public void shouldStayDeadAfterTwoCloses() throws Exception { initializeConsumer(); @@ -209,7 +208,6 @@ public void shouldStayDeadAfterTwoCloses() throws Exception { assertEquals(GlobalStreamThread.State.DEAD, globalStreamThread.state()); } - @SuppressWarnings("unchecked") @Test public void shouldTransitionToRunningOnStart() throws Exception { initializeConsumer(); @@ -224,7 +222,28 @@ public void shouldTransitionToRunningOnStart() throws Exception { } @Test - public void shouldDieOnInvalidOffsetException() throws Exception { + public void shouldDieOnInvalidOffsetExceptionDuringStartup() throws Exception { + initializeConsumer(); + mockConsumer.setPollException(new InvalidOffsetException("Try Again!") { + @Override + public Set partitions() { + return Collections.singleton(topicPartition); + } + }); + + globalStreamThread.start(); + + TestUtils.waitForCondition( + () -> globalStreamThread.state() == DEAD, + 10 * 1000, + "GlobalStreamThread should have died." + ); + + assertFalse(new File(baseDirectoryName + File.separator + "testAppId" + File.separator + "global").exists()); + } + + @Test + public void shouldDieOnInvalidOffsetExceptionWhileRunning() throws Exception { initializeConsumer(); globalStreamThread.start(); @@ -247,13 +266,14 @@ public Set partitions() { return Collections.singleton(topicPartition); } }); - // feed first record for recovery - mockConsumer.addRecord(new ConsumerRecord<>(GLOBAL_STORE_TOPIC_NAME, 0, 0L, "K1".getBytes(), "V1".getBytes())); TestUtils.waitForCondition( () -> globalStreamThread.state() == DEAD, 10 * 1000, - "GlobalStreamThread should have died."); + "GlobalStreamThread should have died." + ); + + assertFalse(new File(baseDirectoryName + File.separator + "testAppId" + File.separator + "global").exists()); } private void initializeConsumer() { diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java index 606b380e5f560..1f98eb456d917 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java @@ -110,21 +110,27 @@ public void shouldNotFlushOffsetsWhenFlushIntervalHasNotLapsed() { @Test public void shouldCloseConsumer() throws IOException { - stateConsumer.close(); + stateConsumer.close(false); assertTrue(consumer.closed()); } @Test public void shouldCloseStateMaintainer() throws IOException { - stateConsumer.close(); + stateConsumer.close(false); assertTrue(stateMaintainer.closed); } + @Test + public void shouldWipeStoreOnClose() throws IOException { + stateConsumer.close(true); + assertTrue(stateMaintainer.wipeStore); + } private static class TaskStub implements GlobalStateMaintainer { private final Map partitionOffsets; private final Map updatedPartitions = new HashMap<>(); private boolean flushed; + private boolean wipeStore; private boolean closed; TaskStub(final Map partitionOffsets) { @@ -141,8 +147,9 @@ public void flushState() { } @Override - public void close() { + public void close(final boolean wipeStateStore) { closed = true; + wipeStore = wipeStateStore; } @Override diff --git a/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java b/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java index 48e7295c6b160..d737d69261991 100644 --- a/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java +++ b/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java @@ -21,22 +21,26 @@ import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.processor.internals.GlobalStateManager; import org.apache.kafka.streams.processor.internals.InternalProcessorContext; +import org.apache.kafka.streams.processor.internals.Task.TaskType; import java.io.File; import java.util.Map; import java.util.Set; -import org.apache.kafka.streams.processor.internals.Task.TaskType; public class GlobalStateManagerStub implements GlobalStateManager { private final Set storeNames; private final Map offsets; + private final File baseDirectory; public boolean initialized; public boolean closed; - public GlobalStateManagerStub(final Set storeNames, final Map offsets) { + public GlobalStateManagerStub(final Set storeNames, + final Map offsets, + final File baseDirectory) { this.storeNames = storeNames; this.offsets = offsets; + this.baseDirectory = baseDirectory; } @Override @@ -50,7 +54,7 @@ public Set initialize() { @Override public File baseDir() { - return null; + return baseDirectory; } @Override diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java index 31659bbeb5514..225fd743f6568 100644 --- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java +++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java @@ -436,11 +436,11 @@ private void setupGlobalTask(final StreamsConfig streamsConfig, globalStateManager.setGlobalProcessorContext(globalProcessorContext); globalStateTask = new GlobalStateUpdateTask( + logContext, globalTopology, globalProcessorContext, globalStateManager, - new LogAndContinueExceptionHandler(), - logContext + new LogAndContinueExceptionHandler() ); globalStateTask.initialize(); globalProcessorContext.setRecordContext(new ProcessorRecordContext( @@ -1181,7 +1181,7 @@ public void close() { } if (globalStateTask != null) { try { - globalStateTask.close(); + globalStateTask.close(false); } catch (final IOException e) { // ignore }