You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by vv...@apache.org on 2020/07/10 02:03:35 UTC
[kafka] branch 2.6 updated: KAFKA-10249: don't try to read
un-checkpointed offsets of in-memory stores (#8996)
This is an automated email from the ASF dual-hosted git repository.
vvcephei pushed a commit to branch 2.6
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.6 by this push:
new 080a331 KAFKA-10249: don't try to read un-checkpointed offsets of in-memory stores (#8996)
080a331 is described below
commit 080a331a65f779a7378156b10ee51400a894c735
Author: A. Sophie Blee-Goldman <so...@confluent.io>
AuthorDate: Thu Jul 9 18:49:20 2020 -0700
KAFKA-10249: don't try to read un-checkpointed offsets of in-memory stores (#8996)
Fixes an asymmetry in which we avoid writing checkpoints for non-persistent stores, but still expected to read them, resulting in a spurious TaskCorruptedException.
Reviewers: Matthias J. Sax <mj...@apache.org>, John Roesler <vv...@apache.org>
---
.../processor/internals/ProcessorStateManager.java | 8 +++
.../processor/internals/StateManagerUtil.java | 21 +++---
.../internals/ProcessorStateManagerTest.java | 79 ++++++++++++++++++++--
.../processor/internals/StateManagerUtilTest.java | 29 ++++++++
4 files changed, 123 insertions(+), 14 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
index 953a4e8..1de599d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
@@ -222,8 +222,16 @@ public class ProcessorStateManager implements StateManager {
log.trace("Loaded offsets from the checkpoint file: {}", loadedCheckpoints);
for (final StateStoreMetadata store : stores.values()) {
+ if (store.corrupted) {
+ log.error("Tried to initialize store offsets for corrupted store {}", store);
+ throw new IllegalStateException("Should not initialize offsets for a corrupted task");
+ }
+
if (store.changelogPartition == null) {
log.info("State store {} is not logged and hence would not be restored", store.stateStore.name());
+ } else if (!store.stateStore.persistent()) {
+ log.info("Initializing to the starting offset for changelog {} of in-memory state store {}",
+ store.changelogPartition, store.stateStore.name());
} else if (store.offset() == null) {
if (loadedCheckpoints.containsKey(store.changelogPartition)) {
final Long offset = changelogOffsetFromCheckpointedOffset(loadedCheckpoints.remove(store.changelogPartition));
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java
index 422b124..57b1b29 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java
@@ -104,18 +104,20 @@ final class StateManagerUtil {
if (stateDirectory.lock(id)) {
try {
stateMgr.close();
-
- if (wipeStateStore) {
- log.debug("Wiping state stores for {} task {}", taskType, id);
- // we can just delete the whole dir of the task, including the state store images and the checkpoint files,
- // and then we write an empty checkpoint file indicating that the previous close is graceful and we just
- // need to re-bootstrap the restoration from the beginning
- Utils.delete(stateMgr.baseDir());
- }
} catch (final ProcessorStateException e) {
firstException.compareAndSet(null, e);
} finally {
- stateDirectory.unlock(id);
+ try {
+ if (wipeStateStore) {
+ log.debug("Wiping state stores for {} task {}", taskType, id);
+ // we can just delete the whole dir of the task, including the state store images and the checkpoint files,
+ // and then we write an empty checkpoint file indicating that the previous close is graceful and we just
+ // need to re-bootstrap the restoration from the beginning
+ Utils.delete(stateMgr.baseDir());
+ }
+ } finally {
+ stateDirectory.unlock(id);
+ }
}
}
} catch (final IOException e) {
@@ -123,7 +125,6 @@ final class StateManagerUtil {
String.format("%sFatal error while trying to close the state manager for task %s", logPrefix, id), e
);
firstException.compareAndSet(null, exception);
-
}
final ProcessorStateException exception = firstException.get();
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
index 77ed3a4..8d4e35b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
@@ -410,14 +410,14 @@ public class ProcessorStateManagerTest {
assertEquals(mkMap(
mkEntry(persistentStorePartition, checkpointOffset + 1L),
mkEntry(persistentStoreTwoPartition, 0L),
- mkEntry(nonPersistentStorePartition, checkpointOffset + 1L)),
+ mkEntry(nonPersistentStorePartition, 0L)),
stateMgr.changelogOffsets()
);
assertNull(stateMgr.storeMetadata(irrelevantPartition));
assertNull(stateMgr.storeMetadata(persistentStoreTwoPartition).offset());
assertThat(stateMgr.storeMetadata(persistentStorePartition).offset(), equalTo(checkpointOffset));
- assertThat(stateMgr.storeMetadata(nonPersistentStorePartition).offset(), equalTo(checkpointOffset));
+ assertNull(stateMgr.storeMetadata(nonPersistentStorePartition).offset());
} finally {
stateMgr.close();
}
@@ -533,7 +533,7 @@ public class ProcessorStateManagerTest {
}
@Test
- public void shouldNotWriteCheckpointForNonPersistent() throws IOException {
+ public void shouldNotWriteCheckpointForNonPersistentStore() throws IOException {
final ProcessorStateManager stateMgr = getStateManager(Task.TaskType.ACTIVE);
try {
@@ -784,7 +784,7 @@ public class ProcessorStateManagerTest {
}
@Test
- public void shouldThrowTaskCorruptedWithoutCheckpointNonEmptyDir() throws IOException {
+ public void shouldThrowTaskCorruptedWithoutPersistentStoreCheckpointAndNonEmptyDir() throws IOException {
final long checkpointOffset = 10L;
final Map<TopicPartition, Long> offsets = mkMap(
@@ -814,6 +814,77 @@ public class ProcessorStateManagerTest {
}
@Test
+ public void shouldNotThrowTaskCorruptedWithoutInMemoryStoreCheckpointAndNonEmptyDir() throws IOException {
+ final long checkpointOffset = 10L;
+
+ final Map<TopicPartition, Long> offsets = mkMap(
+ mkEntry(persistentStorePartition, checkpointOffset),
+ mkEntry(irrelevantPartition, 999L)
+ );
+ checkpoint.write(offsets);
+
+ final ProcessorStateManager stateMgr = getStateManager(Task.TaskType.ACTIVE, true);
+
+ try {
+ stateMgr.registerStore(persistentStore, persistentStore.stateRestoreCallback);
+ stateMgr.registerStore(nonPersistentStore, nonPersistentStore.stateRestoreCallback);
+
+ stateMgr.initializeStoreOffsetsFromCheckpoint(false);
+ } finally {
+ stateMgr.close();
+ }
+ }
+
+ @Test
+ public void shouldNotThrowTaskCorruptedExceptionAfterCheckpointing() {
+ final ProcessorStateManager stateMgr = getStateManager(Task.TaskType.ACTIVE, true);
+
+ try {
+ stateMgr.registerStore(persistentStore, persistentStore.stateRestoreCallback);
+ stateMgr.registerStore(nonPersistentStore, nonPersistentStore.stateRestoreCallback);
+ stateMgr.initializeStoreOffsetsFromCheckpoint(true);
+
+ assertThat(stateMgr.storeMetadata(nonPersistentStorePartition), notNullValue());
+ assertThat(stateMgr.storeMetadata(persistentStorePartition), notNullValue());
+
+ stateMgr.checkpoint(mkMap(
+ mkEntry(nonPersistentStorePartition, 876L),
+ mkEntry(persistentStorePartition, 666L))
+ );
+
+ // reset the state and offsets, for example as in a corrupted task
+ stateMgr.close();
+ assertNull(stateMgr.storeMetadata(nonPersistentStorePartition));
+ assertNull(stateMgr.storeMetadata(persistentStorePartition));
+
+ stateMgr.registerStore(persistentStore, persistentStore.stateRestoreCallback);
+ stateMgr.registerStore(nonPersistentStore, nonPersistentStore.stateRestoreCallback);
+
+ // This should not throw a TaskCorruptedException!
+ stateMgr.initializeStoreOffsetsFromCheckpoint(false);
+ assertThat(stateMgr.storeMetadata(nonPersistentStorePartition), notNullValue());
+ assertThat(stateMgr.storeMetadata(persistentStorePartition), notNullValue());
+ } finally {
+ stateMgr.close();
+ }
+ }
+
+ @Test
+ public void shouldThrowIllegalStateIfInitializingOffsetsForCorruptedTasks() {
+ final ProcessorStateManager stateMgr = getStateManager(Task.TaskType.ACTIVE, true);
+
+ try {
+ stateMgr.registerStore(persistentStore, persistentStore.stateRestoreCallback);
+ stateMgr.markChangelogAsCorrupted(mkSet(persistentStorePartition));
+
+ final ProcessorStateException thrown = assertThrows(ProcessorStateException.class, () -> stateMgr.initializeStoreOffsetsFromCheckpoint(true));
+ assertTrue(thrown.getCause() instanceof IllegalStateException);
+ } finally {
+ stateMgr.close();
+ }
+ }
+
+ @Test
public void shouldBeAbleToCloseWithoutRegisteringAnyStores() {
final ProcessorStateManager stateMgr = getStateManager(Task.TaskType.ACTIVE, true);
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilTest.java
index 9f2a5fd..62ac356 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilTest.java
@@ -288,6 +288,35 @@ public class StateManagerUtilTest {
}
@Test
+ public void shouldStillWipeStateStoresIfCloseThrowsException() throws IOException {
+ final File randomFile = new File("/random/path");
+ mockStatic(Utils.class);
+
+ expect(stateManager.taskId()).andReturn(taskId);
+ expect(stateDirectory.lock(taskId)).andReturn(true);
+
+ stateManager.close();
+ expectLastCall().andThrow(new ProcessorStateException("Close failed"));
+
+ expect(stateManager.baseDir()).andReturn(randomFile);
+
+ Utils.delete(randomFile);
+
+ stateDirectory.unlock(taskId);
+ expectLastCall();
+
+ ctrl.checkOrder(true);
+ ctrl.replay();
+
+ replayAll();
+
+ assertThrows(ProcessorStateException.class, () ->
+ StateManagerUtil.closeStateManager(logger, "logPrefix:", false, true, stateManager, stateDirectory, TaskType.ACTIVE));
+
+ ctrl.verify();
+ }
+
+ @Test
public void testCloseStateManagerWithStateStoreWipeOutRethrowWrappedIOException() throws IOException {
final File unknownFile = new File("/unknown/path");
mockStatic(Utils.class);