[
https://issues.apache.org/jira/browse/KAFKA-6730?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16473238#comment-16473238
]
ASF GitHub Bot commented on KAFKA-6730:
---------------------------------------
ConcurrencyPractitioner closed pull request #4901: [KAFKA-6730] Simplify state
store recovery
URL: https://github.com/apache/kafka/pull/4901
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
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 9b67fc4263c..a20f1acf675 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 @@
private final Map<TopicPartition, StateRestorer> stateRestorers = new
HashMap<>();
private final Map<TopicPartition, StateRestorer> needsRestoring = new
HashMap<>();
private final Map<TopicPartition, StateRestorer> needsInitializing = new
HashMap<>();
+ private boolean hasRetrievedOffsets;
+ private Map<TopicPartition, Long> updatedEndOffsets;
public StoreChangelogReader(final Consumer<byte[], byte[]> restoreConsumer,
final StateRestoreListener
userStateRestoreListener,
@@ -59,6 +61,7 @@ public StoreChangelogReader(final Consumer<byte[], byte[]>
restoreConsumer,
this.restoreConsumer = restoreConsumer;
this.log = logContext.logger(getClass());
this.userStateRestoreListener = userStateRestoreListener;
+ this.hasRetrievedOffsets = false;
}
@Override
@@ -83,10 +86,25 @@ public void register(final StateRestorer restorer) {
final Set<TopicPartition> restoringPartitions = new
HashSet<>(needsRestoring.keySet());
try {
- final ConsumerRecords<byte[], byte[]> 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<byte[], byte[]> records =
poll(restoreConsumer, 10);
+ final Iterator<TopicPartition> iterator =
restoringPartitions.iterator();
+ final Set<TopicPartition> 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<TopicPartition> 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<byte[], byte[]>
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<ConsumerRecord<byte[], byte[]>>
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 6308ca7fd80..b0a78bfa407 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<byte[], byte[]> 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 c65d4efadb1..1739dfce4fb 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;
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Simplify state store recovery
> -----------------------------
>
> Key: KAFKA-6730
> URL: https://issues.apache.org/jira/browse/KAFKA-6730
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Reporter: Matthias J. Sax
> Assignee: Richard Yu
> Priority: Major
> Fix For: 2.0.0
>
>
> In the current code base, we restore state stores in the main thread (in
> contrast to older code that did restore state stored in the rebalance call
> back). This has multiple advantages and allows us the further simplify
> restore code.
> In the original code base, during a long restore phase, it was possible that
> a instance misses a rebalance and drops out of the consumer group. To detect
> this case, we apply a check during the restore phase, that the end-offset of
> the changelog topic does not change. A changed offset implies a missed
> rebalance as another thread started to write into the changelog topic (ie,
> the current thread does not own the task/store/changelog-topic anymore).
> With the new code, that restores in the main-loop, it's ensured that `poll()`
> is called regularly and thus, a rebalance will be detected automatically.
> This make the check about an changing changelog-end-offset unnecessary.
> We can simplify the restore logic, to just consuming until `poll()` does not
> return any data. For this case, we fetch the end-offset to see if we did
> fully restore. If yes, we resume processing, if not, we continue the restore.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
