KAFKA-6782: solved the bug of restoration of aborted messages for Glo…

…balStateStore and KGlobalTable (apache#4900) Reviewer: Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bill@confluent.io>, Guozhang Wang <guozhang@confluent.io>
confluentinc · Jun 12, 2018 · 40f63eb · 40f63eb
1 parent 7a59061
commit 40f63eb
Show file tree

Hide file tree

Showing 6 changed files with 429 additions and 66 deletions.
diff --git a/.gitignore b/.gitignore
@@ -1,5 +1,6 @@
 dist
 *classes
+*.class
 target/
 build/
 build_eclipse/

diff --git a/kafka b/kafka
diff --git a/...ms/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java b/...ms/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
@@ -271,8 +271,8 @@ private void restoreState(final StateRestoreCallback stateRestoreCallback,
                         if (record.key() != null) {
                             restoreRecords.add(KeyValue.pair(record.key(), record.value()));
                         }
-                        offset = globalConsumer.position(topicPartition);
                     }
+                    offset = globalConsumer.position(topicPartition);
                     stateRestoreAdapter.restoreAll(restoreRecords);
                     stateRestoreListener.onBatchRestored(topicPartition, storeName, offset, restoreRecords.size());
                     restoreCount += restoreRecords.size();