KAFKA-10306: GlobalThread should fail on InvalidOffsetException

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<byte[], byte[]> record);
 }

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<byte[], byte[]> records = globalConsumer.poll(pollTime);
-                    final List<ConsumerRecord<byte[], byte[]>> restoreRecords = new ArrayList<>();
-                    for (final ConsumerRecord<byte[], byte[]> record : records.records(topicPartition)) {
-                        if (record.key() != null) {
-                            restoreRecords.add(recordConverter.convert(record));
-                        }
+                final ConsumerRecords<byte[], byte[]> records = globalConsumer.poll(pollTime);
+                final List<ConsumerRecord<byte[], byte[]>> restoreRecords = new ArrayList<>();
+                for (final ConsumerRecord<byte[], byte[]> 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);

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,25 +35,27 @@
  * 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;
     private final Map<TopicPartition, Long> offsets = new HashMap<>();
     private final Map<String, RecordDeserializer> 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() {

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java

@@ -234,24 +234,18 @@ void initialize() {
         }
 
         void pollAndUpdate() {
-            try {
-                final ConsumerRecords<byte[], byte[]> received = globalConsumer.poll(pollTime);
-                for (final ConsumerRecord<byte[], byte[]> 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<byte[], byte[]> received = globalConsumer.poll(pollTime);
+            for (final ConsumerRecord<byte[], byte[]> 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. 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) {

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<TopicPartition> partitions() {
-                return Collections.singleton(t1);
-            }
-        });
-
-        stateManager.initialize();
-
-        stateManager.registerStore(store1, stateRestoreCallback);
-        assertEquals(2, stateRestoreCallback.restored.size());
-    }
-
     @Test
     public void shouldListenForRestoreEvents() {
         initializeConsumer(5, 1, t1);

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<TopicPartition, Long> 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());
+    }
 }