From 9b41afa5f1bc38d52c3aa5f2424c3d5ee1f77ffd Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Sat, 12 May 2018 11:32:08 -0700 Subject: [PATCH 01/10] [KAFKA-6730] Simplify State Store Recovery --- .../internals/StoreChangelogReader.java | 60 +++++++------------ .../processor/internals/TaskManager.java | 1 - .../internals/StoreChangelogReaderTest.java | 4 ++ 3 files changed, 26 insertions(+), 39 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index 9b67fc4263c7e..a20f1acf67547 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -52,6 +52,8 @@ public class StoreChangelogReader implements ChangelogReader { private final Map stateRestorers = new HashMap<>(); private final Map needsRestoring = new HashMap<>(); private final Map needsInitializing = new HashMap<>(); + private boolean hasRetrievedOffsets; + private Map updatedEndOffsets; public StoreChangelogReader(final Consumer restoreConsumer, final StateRestoreListener userStateRestoreListener, @@ -59,6 +61,7 @@ public StoreChangelogReader(final Consumer restoreConsumer, this.restoreConsumer = restoreConsumer; this.log = logContext.logger(getClass()); this.userStateRestoreListener = userStateRestoreListener; + this.hasRetrievedOffsets = false; } @Override @@ -83,10 +86,25 @@ public Collection restore(final RestoringTasks active) { final Set restoringPartitions = new HashSet<>(needsRestoring.keySet()); try { - final ConsumerRecords allRecords = poll(restoreConsumer, 10); - for (final TopicPartition partition : restoringPartitions) { - restorePartition(allRecords, partition, active.restoringTaskFor(partition)); + if (!needsRestoring.isEmpty() && !hasRetrievedOffsets) { + hasRetrievedOffsets = true; + updatedEndOffsets = restoreConsumer.endOffsets(restoringPartitions); } + final ConsumerRecords records = poll(restoreConsumer, 10); + final Iterator iterator = restoringPartitions.iterator(); + final Set completedPartitions = new HashSet<>(); + while (iterator.hasNext()) { + final TopicPartition partition = iterator.next(); + final StateRestorer restorer = stateRestorers.get(partition); + final long pos = processNext(records.records(partition), restorer, updatedEndOffsets.get(partition)); + restorer.setRestoredOffset(pos); + if (restorer.hasCompleted(pos, updatedEndOffsets.get(partition))) { + restorer.restoreDone(); + needsRestoring.remove(partition); + completedPartitions.add(partition); + } + } + restoringPartitions.removeAll(completedPartitions); } catch (final InvalidOffsetException recoverableException) { log.warn("Restoring StreamTasks failed. Deleting StreamTasks stores to recreate from scratch.", recoverableException); final Set partitions = recoverableException.partitions(); @@ -240,41 +258,7 @@ public void reset() { needsRestoring.clear(); endOffsets.clear(); needsInitializing.clear(); - } - - /** - * @throws TaskMigratedException if another thread wrote to the changelog topic that is currently restored - */ - private void restorePartition(final ConsumerRecords allRecords, - final TopicPartition topicPartition, - final Task task) { - final StateRestorer restorer = stateRestorers.get(topicPartition); - final Long endOffset = endOffsets.get(topicPartition); - final long pos = processNext(allRecords.records(topicPartition), restorer, endOffset); - restorer.setRestoredOffset(pos); - if (restorer.hasCompleted(pos, endOffset)) { - if (pos > endOffset) { - throw new TaskMigratedException(task, topicPartition, endOffset, pos); - } - - // need to check for changelog topic - if (restorer.offsetLimit() == Long.MAX_VALUE) { - final Long updatedEndOffset = restoreConsumer.endOffsets(Collections.singletonList(topicPartition)).get(topicPartition); - if (!restorer.hasCompleted(pos, updatedEndOffset)) { - throw new TaskMigratedException(task, topicPartition, updatedEndOffset, pos); - } - } - - - log.debug("Completed restoring state from changelog {} with {} records ranging from offset {} to {}", - topicPartition, - restorer.restoredNumRecords(), - restorer.startingOffset(), - restorer.restoredOffset()); - - restorer.restoreDone(); - needsRestoring.remove(topicPartition); - } + hasRetrievedOffsets = false; } private long processNext(final List> records, diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java index 63224dbcde743..2c06174218ae2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java @@ -309,7 +309,6 @@ void setConsumer(final Consumer consumer) { /** * @throws IllegalStateException If store gets registered after initialized is already finished * @throws StreamsException if the store's change log does not contain the partition - * @throws TaskMigratedException if the task producer got fenced or consumer discovered changelog offset changes (EOS only) */ boolean updateNewAndRestoringTasks() { active.initializeNewTasks(); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java index c65d4efadb14a..1739dfce4fb5e 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java @@ -37,6 +37,7 @@ import org.easymock.MockType; import org.hamcrest.CoreMatchers; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; @@ -377,6 +378,7 @@ public void shouldRestorePartitionsRegisteredPostInitialization() { assertThat(callbackTwo.restored.size(), equalTo(3)); } + @Ignore @Test public void shouldThrowTaskMigratedExceptionIfEndOffsetGetsExceededDuringRestoreForChangelogTopic() { final int messages = 10; @@ -396,6 +398,7 @@ public void shouldThrowTaskMigratedExceptionIfEndOffsetGetsExceededDuringRestore } + @Ignore @Test public void shouldThrowTaskMigratedExceptionIfChangelogTopicUpdatedDuringRestoreProcessFoundInSecondCheck() { final int messages = 10; @@ -434,6 +437,7 @@ public void shouldNotThrowTaskMigratedExceptionIfSourceTopicUpdatedDuringRestore } + @Ignore @Test public void shouldThrowTaskMigratedExceptionIfEndOffsetGetsExceededDuringRestoreForChangelogTopicEOSEnabled() { final int totalMessages = 10; From a129b4e57ecf20053b8d4cf4c47b7223a6494ca5 Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Sat, 12 May 2018 13:21:30 -0700 Subject: [PATCH 02/10] Making change --- .../streams/processor/internals/StoreChangelogReader.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index a20f1acf67547..f791f969d2bc1 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -52,7 +52,6 @@ public class StoreChangelogReader implements ChangelogReader { private final Map stateRestorers = new HashMap<>(); private final Map needsRestoring = new HashMap<>(); private final Map needsInitializing = new HashMap<>(); - private boolean hasRetrievedOffsets; private Map updatedEndOffsets; public StoreChangelogReader(final Consumer restoreConsumer, @@ -61,7 +60,6 @@ public StoreChangelogReader(final Consumer restoreConsumer, this.restoreConsumer = restoreConsumer; this.log = logContext.logger(getClass()); this.userStateRestoreListener = userStateRestoreListener; - this.hasRetrievedOffsets = false; } @Override @@ -86,8 +84,7 @@ public Collection restore(final RestoringTasks active) { final Set restoringPartitions = new HashSet<>(needsRestoring.keySet()); try { - if (!needsRestoring.isEmpty() && !hasRetrievedOffsets) { - hasRetrievedOffsets = true; + if (!needsRestoring.isEmpty()) { updatedEndOffsets = restoreConsumer.endOffsets(restoringPartitions); } final ConsumerRecords records = poll(restoreConsumer, 10); @@ -258,7 +255,6 @@ public void reset() { needsRestoring.clear(); endOffsets.clear(); needsInitializing.clear(); - hasRetrievedOffsets = false; } private long processNext(final List> records, From 4883a431102c52d53d2401bf008e38cde3abe948 Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Wed, 16 May 2018 15:39:04 -0700 Subject: [PATCH 03/10] ensuring updatedEndOffsets is called once --- .../internals/StoreChangelogReader.java | 8 +-- .../internals/StoreChangelogReaderTest.java | 69 ------------------- 2 files changed, 2 insertions(+), 75 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index f791f969d2bc1..74ddf6c02f261 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -26,7 +26,6 @@ import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.errors.StreamsException; -import org.apache.kafka.streams.errors.TaskMigratedException; import org.apache.kafka.streams.processor.StateRestoreListener; import org.slf4j.Logger; @@ -52,7 +51,7 @@ public class StoreChangelogReader implements ChangelogReader { private final Map stateRestorers = new HashMap<>(); private final Map needsRestoring = new HashMap<>(); private final Map needsInitializing = new HashMap<>(); - private Map updatedEndOffsets; + private Map updatedEndOffsets = null; public StoreChangelogReader(final Consumer restoreConsumer, final StateRestoreListener userStateRestoreListener, @@ -69,9 +68,6 @@ public void register(final StateRestorer restorer) { needsInitializing.put(restorer.partition(), restorer); } - /** - * @throws TaskMigratedException if another thread wrote to the changelog topic that is currently restored - */ public Collection restore(final RestoringTasks active) { if (!needsInitializing.isEmpty()) { initialize(); @@ -84,7 +80,7 @@ public Collection restore(final RestoringTasks active) { final Set restoringPartitions = new HashSet<>(needsRestoring.keySet()); try { - if (!needsRestoring.isEmpty()) { + if (!needsRestoring.isEmpty() && updatedEndOffsets == null) { updatedEndOffsets = restoreConsumer.endOffsets(restoringPartitions); } final ConsumerRecords records = poll(restoreConsumer, 10); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java index 1739dfce4fb5e..aabe7ff631362 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java @@ -27,7 +27,6 @@ import org.apache.kafka.common.utils.Utils; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.errors.StreamsException; -import org.apache.kafka.streams.errors.TaskMigratedException; import org.apache.kafka.streams.processor.StateRestoreListener; import org.apache.kafka.test.MockRestoreCallback; import org.apache.kafka.test.MockStateRestoreListener; @@ -37,7 +36,6 @@ import org.easymock.MockType; import org.hamcrest.CoreMatchers; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; @@ -378,48 +376,6 @@ public void shouldRestorePartitionsRegisteredPostInitialization() { assertThat(callbackTwo.restored.size(), equalTo(3)); } - @Ignore - @Test - public void shouldThrowTaskMigratedExceptionIfEndOffsetGetsExceededDuringRestoreForChangelogTopic() { - final int messages = 10; - setupConsumer(messages, topicPartition); - consumer.updateEndOffsets(Collections.singletonMap(topicPartition, 5L)); - changelogReader.register(new StateRestorer(topicPartition, restoreListener, null, Long.MAX_VALUE, true, "storeName")); - - expect(active.restoringTaskFor(topicPartition)).andReturn(task); - replay(active); - - try { - changelogReader.restore(active); - fail("Should have thrown TaskMigratedException"); - } catch (final TaskMigratedException expected) { - /* ignore */ - } - } - - - @Ignore - @Test - public void shouldThrowTaskMigratedExceptionIfChangelogTopicUpdatedDuringRestoreProcessFoundInSecondCheck() { - final int messages = 10; - setupConsumer(messages, topicPartition); - // in this case first call to endOffsets returns correct value, but a second thread has updated the changelog topic - // so a subsequent call to endOffsets returns a value exceeding the expected end value - consumer.addEndOffsets(Collections.singletonMap(topicPartition, 15L)); - changelogReader.register(new StateRestorer(topicPartition, restoreListener, null, Long.MAX_VALUE, true, "storeName")); - - expect(active.restoringTaskFor(topicPartition)).andReturn(task); - replay(active); - - try { - changelogReader.restore(active); - fail("Should have thrown TaskMigratedException"); - } catch (final TaskMigratedException expected) { - // verifies second block threw exception with updated end offset - assertTrue(expected.getMessage().contains("end offset 15, current offset 10")); - } - } - @Test public void shouldNotThrowTaskMigratedExceptionIfSourceTopicUpdatedDuringRestoreProcess() { @@ -437,31 +393,6 @@ public void shouldNotThrowTaskMigratedExceptionIfSourceTopicUpdatedDuringRestore } - @Ignore - @Test - public void shouldThrowTaskMigratedExceptionIfEndOffsetGetsExceededDuringRestoreForChangelogTopicEOSEnabled() { - final int totalMessages = 10; - assignPartition(totalMessages, topicPartition); - // records 0..4 - addRecords(5, topicPartition, 0); - //EOS enabled commit marker at offset 5 so rest of records 6..10 - addRecords(5, topicPartition, 6); - consumer.assign(Collections.emptyList()); - - // end offsets should start after commit marker of 5 from above - consumer.updateEndOffsets(Collections.singletonMap(topicPartition, 6L)); - changelogReader.register(new StateRestorer(topicPartition, restoreListener, null, Long.MAX_VALUE, true, "storeName")); - - expect(active.restoringTaskFor(topicPartition)).andReturn(task); - replay(active); - try { - changelogReader.restore(active); - fail("Should have thrown task migrated exception"); - } catch (final TaskMigratedException expected) { - /* ignore */ - } - } - @Test public void shouldNotThrowTaskMigratedExceptionDuringRestoreForChangelogTopicWhenEndOffsetNotExceededEOSEnabled() { final int totalMessages = 10; From 804a85003881cdebb1e53c183263ef0350c8d3ff Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Sat, 19 May 2018 16:04:19 -0700 Subject: [PATCH 04/10] Fixing post initialization --- .../streams/processor/internals/StoreChangelogReader.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index 74ddf6c02f261..8063c23252a0d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -51,7 +51,7 @@ public class StoreChangelogReader implements ChangelogReader { private final Map stateRestorers = new HashMap<>(); private final Map needsRestoring = new HashMap<>(); private final Map needsInitializing = new HashMap<>(); - private Map updatedEndOffsets = null; + private Map updatedEndOffsets = new HashMap<>(); public StoreChangelogReader(final Consumer restoreConsumer, final StateRestoreListener userStateRestoreListener, @@ -80,8 +80,10 @@ public Collection restore(final RestoringTasks active) { final Set restoringPartitions = new HashSet<>(needsRestoring.keySet()); try { - if (!needsRestoring.isEmpty() && updatedEndOffsets == null) { - updatedEndOffsets = restoreConsumer.endOffsets(restoringPartitions); + if (!needsRestoring.isEmpty()) { + final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); + remainingPartitions.removeAll(updatedEndOffsets.keySet()); + updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); } final ConsumerRecords records = poll(restoreConsumer, 10); final Iterator iterator = restoringPartitions.iterator(); From 78d81c3ef52142e04801212345337eef10086d75 Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Mon, 21 May 2018 19:18:45 -0700 Subject: [PATCH 05/10] Adding removal of partitions --- .../streams/processor/internals/StoreChangelogReader.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index 10608c2f03d0c..b86f2f162bdb3 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -83,7 +83,7 @@ public Collection restore(final RestoringTasks active) { remainingPartitions.removeAll(updatedEndOffsets.keySet()); updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); } - final ConsumerRecords records = poll(restoreConsumer, 10); + final ConsumerRecords records = restoreConsumer.poll(10); final Iterator iterator = restoringPartitions.iterator(); final Set completedPartitions = new HashSet<>(); while (iterator.hasNext()) { @@ -94,6 +94,7 @@ public Collection restore(final RestoringTasks active) { if (restorer.hasCompleted(pos, updatedEndOffsets.get(partition))) { restorer.restoreDone(); needsRestoring.remove(partition); + updatedEndOffsets.remove(partition); completedPartitions.add(partition); } } From 8207dbecd1d4e382c37c14228fef4ff2bb8579b3 Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Sat, 26 May 2018 13:58:57 -0700 Subject: [PATCH 06/10] Removing conditional --- .../processor/internals/StoreChangelogReader.java | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index b86f2f162bdb3..0e7978681b7cb 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -78,11 +78,9 @@ public Collection restore(final RestoringTasks active) { final Set restoringPartitions = new HashSet<>(needsRestoring.keySet()); try { - if (!needsRestoring.isEmpty()) { - final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); - remainingPartitions.removeAll(updatedEndOffsets.keySet()); - updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); - } + final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); + remainingPartitions.removeAll(updatedEndOffsets.keySet()); + updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); final ConsumerRecords records = restoreConsumer.poll(10); final Iterator iterator = restoringPartitions.iterator(); final Set completedPartitions = new HashSet<>(); @@ -305,7 +303,6 @@ private boolean hasPartition(final TopicPartition topicPartition) { return true; } } - return false; } } From 303fbab7e6a5ede5425254d259879d7eacf0edea Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Tue, 29 May 2018 14:35:44 -0700 Subject: [PATCH 07/10] Patching --- .../processor/internals/StoreChangelogReader.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index 0e7978681b7cb..d0522aeb5c120 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -76,14 +76,13 @@ public Collection restore(final RestoringTasks active) { return completed(); } - final Set restoringPartitions = new HashSet<>(needsRestoring.keySet()); try { final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); remainingPartitions.removeAll(updatedEndOffsets.keySet()); updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); final ConsumerRecords records = restoreConsumer.poll(10); - final Iterator iterator = restoringPartitions.iterator(); - final Set completedPartitions = new HashSet<>(); + final Iterator iterator = needsRestoring.keySet().iterator(); + final HashSet completedPartitions = new HashSet<>(); while (iterator.hasNext()) { final TopicPartition partition = iterator.next(); final StateRestorer restorer = stateRestorers.get(partition); @@ -91,12 +90,13 @@ public Collection restore(final RestoringTasks active) { restorer.setRestoredOffset(pos); if (restorer.hasCompleted(pos, updatedEndOffsets.get(partition))) { restorer.restoreDone(); - needsRestoring.remove(partition); updatedEndOffsets.remove(partition); completedPartitions.add(partition); } } - restoringPartitions.removeAll(completedPartitions); + for (TopicPartition partition : completedPartitions) { + needsRestoring.remove(partition); + } } catch (final InvalidOffsetException recoverableException) { log.warn("Restoring StreamTasks failed. Deleting StreamTasks stores to recreate from scratch.", recoverableException); final Set partitions = recoverableException.partitions(); From 4a33d9c3b078f14dcd0dc511046ace188f7ebeb1 Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Tue, 29 May 2018 14:40:42 -0700 Subject: [PATCH 08/10] Strengthening conditional --- .../streams/processor/internals/StoreChangelogReader.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index d0522aeb5c120..05cfdd253ecf6 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -68,6 +68,9 @@ public void register(final StateRestorer restorer) { public Collection restore(final RestoringTasks active) { if (!needsInitializing.isEmpty()) { + final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); + remainingPartitions.removeAll(updatedEndOffsets.keySet()); + updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); initialize(); } @@ -77,9 +80,6 @@ public Collection restore(final RestoringTasks active) { } try { - final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); - remainingPartitions.removeAll(updatedEndOffsets.keySet()); - updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); final ConsumerRecords records = restoreConsumer.poll(10); final Iterator iterator = needsRestoring.keySet().iterator(); final HashSet completedPartitions = new HashSet<>(); From dcb5f0c558c5de20f9cc503e783400f5843d999f Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Thu, 31 May 2018 19:37:14 -0700 Subject: [PATCH 09/10] Reverting previous unforeseen error --- .../processor/internals/StoreChangelogReader.java | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index 05cfdd253ecf6..cb8cfc2a400cf 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -68,9 +68,6 @@ public void register(final StateRestorer restorer) { public Collection restore(final RestoringTasks active) { if (!needsInitializing.isEmpty()) { - final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); - remainingPartitions.removeAll(updatedEndOffsets.keySet()); - updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); initialize(); } @@ -80,9 +77,11 @@ public Collection restore(final RestoringTasks active) { } try { + final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); + remainingPartitions.removeAll(updatedEndOffsets.keySet()); + updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); final ConsumerRecords records = restoreConsumer.poll(10); final Iterator iterator = needsRestoring.keySet().iterator(); - final HashSet completedPartitions = new HashSet<>(); while (iterator.hasNext()) { final TopicPartition partition = iterator.next(); final StateRestorer restorer = stateRestorers.get(partition); @@ -91,12 +90,9 @@ public Collection restore(final RestoringTasks active) { if (restorer.hasCompleted(pos, updatedEndOffsets.get(partition))) { restorer.restoreDone(); updatedEndOffsets.remove(partition); - completedPartitions.add(partition); + iterator.remove(); } } - for (TopicPartition partition : completedPartitions) { - needsRestoring.remove(partition); - } } catch (final InvalidOffsetException recoverableException) { log.warn("Restoring StreamTasks failed. Deleting StreamTasks stores to recreate from scratch.", recoverableException); final Set partitions = recoverableException.partitions(); From e742d310817ba306c0a217666b983755418549e1 Mon Sep 17 00:00:00 2001 From: Richard Yu Date: Mon, 4 Jun 2018 14:37:18 -0700 Subject: [PATCH 10/10] Moving partition addition --- .../streams/processor/internals/StoreChangelogReader.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index cb8cfc2a400cf..af5ff4728f5e3 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -69,6 +69,9 @@ public void register(final StateRestorer restorer) { public Collection restore(final RestoringTasks active) { if (!needsInitializing.isEmpty()) { initialize(); + final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); + remainingPartitions.removeAll(updatedEndOffsets.keySet()); + updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); } if (needsRestoring.isEmpty()) { @@ -77,9 +80,6 @@ public Collection restore(final RestoringTasks active) { } try { - final Set remainingPartitions = new HashSet<>(needsRestoring.keySet()); - remainingPartitions.removeAll(updatedEndOffsets.keySet()); - updatedEndOffsets.putAll(restoreConsumer.endOffsets(remainingPartitions)); final ConsumerRecords records = restoreConsumer.poll(10); final Iterator iterator = needsRestoring.keySet().iterator(); while (iterator.hasNext()) {