kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mj...@apache.org
Subject [kafka] branch 2.1 updated: KAFKA-7443: OffsetOutOfRangeException in restoring state store from changelog topic when start offset of local checkpoint is smaller than that of changelog topic (#5946)
Date Tue, 11 Dec 2018 08:48:24 GMT
This is an automated email from the ASF dual-hosted git repository.

mjsax pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.1 by this push:
     new 8157e98  KAFKA-7443: OffsetOutOfRangeException in restoring state store from changelog
topic when start offset of local checkpoint is smaller than that of changelog topic (#5946)
8157e98 is described below

commit 8157e984e92161c4e68678f158d05837edac2e62
Author: linyli001 <45358531+linyli001@users.noreply.github.com>
AuthorDate: Tue Dec 11 16:40:18 2018 +0800

    KAFKA-7443: OffsetOutOfRangeException in restoring state store from changelog topic when
start offset of local checkpoint is smaller than that of changelog topic (#5946)
    
    Reviewer: Matthias J. Sax <matthias@confluent.io>, John Roesler <john@confluent.io>
---
 .../kafka/clients/consumer/MockConsumer.java       |  4 +++
 .../processor/internals/StoreChangelogReader.java  |  2 ++
 .../internals/StoreChangelogReaderTest.java        | 36 ++++++++++++++++++++++
 3 files changed, 42 insertions(+)

diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
index 9eee6da..4af3ba1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
@@ -191,6 +191,10 @@ public class MockConsumer<K, V> implements Consumer<K, V>
{
             if (!subscriptions.isPaused(entry.getKey())) {
                 final List<ConsumerRecord<K, V>> recs = entry.getValue();
                 for (final ConsumerRecord<K, V> rec : recs) {
+                    if (beginningOffsets.get(entry.getKey()) != null && beginningOffsets.get(entry.getKey())
> subscriptions.position(entry.getKey())) {
+                        throw new OffsetOutOfRangeException(Collections.singletonMap(entry.getKey(),
subscriptions.position(entry.getKey())));
+                    }
+
                     if (assignment().contains(entry.getKey()) && rec.offset() >=
subscriptions.position(entry.getKey())) {
                         results.get(entry.getKey()).add(rec);
                         subscriptions.position(entry.getKey(), rec.offset() + 1);
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 34e6e5c..fdd9d6c 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
@@ -107,6 +107,8 @@ public class StoreChangelogReader implements ChangelogReader {
                 needsInitializing.remove(partition);
                 needsRestoring.remove(partition);
 
+                final StateRestorer restorer = stateRestorers.get(partition);
+                restorer.setCheckpointOffset(StateRestorer.NO_CHECKPOINT);
                 task.reinitializeStateStoresForPartitions(recoverableException.partitions());
             }
             restoreConsumer.seekToBeginning(partitions);
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 ae48f57..aa6395b 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
@@ -156,6 +156,42 @@ public class StoreChangelogReaderTest {
         assertThat(callback.restored.size(), equalTo(messages));
     }
 
+    @Test
+    public void shouldRecoverFromOffsetOutOfRangeExceptionAndRestoreFromStart() {
+        final int messages = 10;
+        final int startOffset = 5;
+        final long expiredCheckpoint = 1L;
+        assignPartition(messages, topicPartition);
+        consumer.updateBeginningOffsets(Collections.singletonMap(topicPartition, (long) startOffset));
+        consumer.updateEndOffsets(Collections.singletonMap(topicPartition, (long) (messages
+ startOffset)));
+
+        addRecords(messages, topicPartition, startOffset);
+        consumer.assign(Collections.<TopicPartition>emptyList());
+
+        final StateRestorer stateRestorer = new StateRestorer(
+                topicPartition,
+                restoreListener,
+                expiredCheckpoint,
+                Long.MAX_VALUE,
+                true,
+                "storeName");
+        changelogReader.register(stateRestorer);
+
+        EasyMock.expect(active.restoringTaskFor(topicPartition)).andStubReturn(task);
+        EasyMock.replay(active, task);
+
+        // first restore call "fails" since OffsetOutOfRangeException but we should not die
with an exception
+        assertEquals(0, changelogReader.restore(active).size());
+        //the starting offset for stateRestorer is set to NO_CHECKPOINT
+        assertThat(stateRestorer.checkpoint(), equalTo(-1L));
+
+        //restore the active task again
+        changelogReader.register(stateRestorer);
+        //the restored task should return completed partition without Exception.
+        assertEquals(1, changelogReader.restore(active).size());
+        //the restored size should be equal to message length.
+        assertThat(callback.restored.size(), equalTo(messages));
+    }
 
     @Test
     public void shouldRestoreMessagesFromCheckpoint() {


Mime
View raw message