You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2017/12/11 14:22:42 UTC
[1/2] kafka git commit: KAFKA-6121: Restore and global consumer
should not use auto.offset.reset
Repository: kafka
Updated Branches:
refs/heads/trunk 2bf2348b5 -> 043951753
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
index 20cf125..df8d201 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.PartitionInfo;
@@ -33,8 +34,8 @@ import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.StateRestoreCallback;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
+import org.apache.kafka.test.MockProcessorContext;
import org.apache.kafka.test.MockStateRestoreListener;
-import org.apache.kafka.test.NoOpProcessorContext;
import org.apache.kafka.test.NoOpReadOnlyStore;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
@@ -70,45 +71,57 @@ public class GlobalStateManagerImplTest {
private final MockTime time = new MockTime();
private final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
private final MockStateRestoreListener stateRestoreListener = new MockStateRestoreListener();
+ private final String storeName1 = "t1-store";
+ private final String storeName2 = "t2-store";
+ private final String storeName3 = "t3-store";
+ private final String storeName4 = "t4-store";
private final TopicPartition t1 = new TopicPartition("t1", 1);
private final TopicPartition t2 = new TopicPartition("t2", 1);
+ private final TopicPartition t3 = new TopicPartition("t3", 1);
+ private final TopicPartition t4 = new TopicPartition("t4", 1);
private GlobalStateManagerImpl stateManager;
- private NoOpProcessorContext context;
private StateDirectory stateDirectory;
- private StreamsConfig config;
- private NoOpReadOnlyStore<Object, Object> store1;
- private NoOpReadOnlyStore store2;
+ private StreamsConfig streamsConfig;
+ private NoOpReadOnlyStore<Object, Object> store1, store2, store3, store4;
private MockConsumer<byte[], byte[]> consumer;
private File checkpointFile;
private ProcessorTopology topology;
+ private MockProcessorContext mockProcessorContext;
@Before
public void before() throws IOException {
final Map<String, String> storeToTopic = new HashMap<>();
- store1 = new NoOpReadOnlyStore<>("t1-store");
- store2 = new NoOpReadOnlyStore("t2-store");
- storeToTopic.put("t1-store", "t1");
- storeToTopic.put("t2-store", "t2");
- topology = ProcessorTopology.withGlobalStores(Utils.<StateStore>mkList(store1, store2), storeToTopic);
+ storeToTopic.put(storeName1, t1.topic());
+ storeToTopic.put(storeName2, t2.topic());
+ storeToTopic.put(storeName3, t3.topic());
+ storeToTopic.put(storeName4, t4.topic());
- context = new NoOpProcessorContext();
- config = new StreamsConfig(new Properties() {
+ store1 = new NoOpReadOnlyStore<>(storeName1, true);
+ store2 = new NoOpReadOnlyStore<>(storeName2, true);
+ store3 = new NoOpReadOnlyStore<>(storeName3);
+ store4 = new NoOpReadOnlyStore<>(storeName4);
+
+ topology = ProcessorTopology.withGlobalStores(Utils.<StateStore>mkList(store1, store2, store3, store4), storeToTopic);
+
+ streamsConfig = new StreamsConfig(new Properties() {
{
put(StreamsConfig.APPLICATION_ID_CONFIG, "appId");
put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
}
});
- stateDirectory = new StateDirectory(config, time);
- consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+ stateDirectory = new StateDirectory(streamsConfig, time);
+ consumer = new MockConsumer<>(OffsetResetStrategy.NONE);
stateManager = new GlobalStateManagerImpl(
- new LogContext("mock"),
+ new LogContext("test"),
topology,
consumer,
stateDirectory,
stateRestoreListener,
- config);
+ streamsConfig);
+ mockProcessorContext = new MockProcessorContext(stateDirectory.globalStateDir(), streamsConfig);
+ stateManager.setGlobalProcessorContext(mockProcessorContext);
checkpointFile = new File(stateManager.baseDir(), ProcessorStateManager.CHECKPOINT_FILE_NAME);
}
@@ -119,16 +132,16 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldLockGlobalStateDirectory() {
- stateManager.initialize(context);
+ stateManager.initialize();
assertTrue(new File(stateDirectory.globalStateDir(), ".lock").exists());
}
@Test(expected = LockException.class)
public void shouldThrowLockExceptionIfCantGetLock() throws IOException {
- final StateDirectory stateDir = new StateDirectory(config, time);
+ final StateDirectory stateDir = new StateDirectory(streamsConfig, time);
try {
stateDir.lockGlobalState();
- stateManager.initialize(context);
+ stateManager.initialize();
} finally {
stateDir.unlockGlobalState();
}
@@ -138,7 +151,7 @@ public class GlobalStateManagerImplTest {
public void shouldReadCheckpointOffsets() throws IOException {
final Map<TopicPartition, Long> expected = writeCheckpoint();
- stateManager.initialize(context);
+ stateManager.initialize();
final Map<TopicPartition, Long> offsets = stateManager.checkpointed();
assertEquals(expected, offsets);
}
@@ -146,35 +159,35 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldNotDeleteCheckpointFileAfterLoaded() throws IOException {
writeCheckpoint();
- stateManager.initialize(context);
+ stateManager.initialize();
assertTrue(checkpointFile.exists());
}
@Test(expected = StreamsException.class)
public void shouldThrowStreamsExceptionIfFailedToReadCheckpointedOffsets() throws IOException {
writeCorruptCheckpoint();
- stateManager.initialize(context);
+ stateManager.initialize();
}
@Test
public void shouldInitializeStateStores() {
- stateManager.initialize(context);
+ stateManager.initialize();
assertTrue(store1.initialized);
assertTrue(store2.initialized);
}
@Test
public void shouldReturnInitializedStoreNames() {
- final Set<String> storeNames = stateManager.initialize(context);
- assertEquals(Utils.mkSet(store1.name(), store2.name()), storeNames);
+ final Set<String> storeNames = stateManager.initialize();
+ assertEquals(Utils.mkSet(storeName1, storeName2, storeName3, storeName4), storeNames);
}
@Test
public void shouldThrowIllegalArgumentIfTryingToRegisterStoreThatIsNotGlobal() {
- stateManager.initialize(context);
+ stateManager.initialize();
try {
- stateManager.register(new NoOpReadOnlyStore<>("not-in-topology"), new TheStateRestoreCallback());
+ stateManager.register(new NoOpReadOnlyStore<>("not-in-topology"), stateRestoreCallback);
fail("should have raised an illegal argument exception as store is not in the topology");
} catch (final IllegalArgumentException e) {
// pass
@@ -183,11 +196,11 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldThrowIllegalArgumentExceptionIfAttemptingToRegisterStoreTwice() {
- stateManager.initialize(context);
+ stateManager.initialize();
initializeConsumer(2, 1, t1);
- stateManager.register(store1, new TheStateRestoreCallback());
+ stateManager.register(store1, stateRestoreCallback);
try {
- stateManager.register(store1, new TheStateRestoreCallback());
+ stateManager.register(store1, stateRestoreCallback);
fail("should have raised an illegal argument exception as store has already been registered");
} catch (final IllegalArgumentException e) {
// pass
@@ -196,9 +209,9 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldThrowStreamsExceptionIfNoPartitionsFoundForStore() {
- stateManager.initialize(context);
+ stateManager.initialize();
try {
- stateManager.register(store1, new TheStateRestoreCallback());
+ stateManager.register(store1, stateRestoreCallback);
fail("Should have raised a StreamsException as there are no partition for the store");
} catch (final StreamsException e) {
// pass
@@ -209,9 +222,23 @@ public class GlobalStateManagerImplTest {
public void shouldRestoreRecordsUpToHighwatermark() {
initializeConsumer(2, 1, t1);
- stateManager.initialize(context);
+ stateManager.initialize();
+
+ stateManager.register(store1, stateRestoreCallback);
+ assertEquals(2, stateRestoreCallback.restored.size());
+ }
+
+ @Test
+ public void shouldRecoverFromInvalidOffsetExceptionAndRestoreRecords() {
+ initializeConsumer(2, 1, t1);
+ consumer.setException(new InvalidOffsetException("Try Again!") {
+ public Set<TopicPartition> partitions() {
+ return Collections.singleton(t1);
+ }
+ });
+
+ stateManager.initialize();
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
stateManager.register(store1, stateRestoreCallback);
assertEquals(2, stateRestoreCallback.restored.size());
}
@@ -219,9 +246,8 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldListenForRestoreEvents() {
initializeConsumer(5, 1, t1);
- stateManager.initialize(context);
+ stateManager.initialize();
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
stateManager.register(store1, stateRestoreCallback);
assertThat(stateRestoreListener.restoreStartOffset, equalTo(1L));
@@ -242,8 +268,7 @@ public class GlobalStateManagerImplTest {
ProcessorStateManager.CHECKPOINT_FILE_NAME));
offsetCheckpoint.write(Collections.singletonMap(t1, 6L));
- stateManager.initialize(context);
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+ stateManager.initialize();
stateManager.register(store1, stateRestoreCallback);
assertEquals(5, stateRestoreCallback.restored.size());
}
@@ -251,8 +276,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldFlushStateStores() {
- stateManager.initialize(context);
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+ stateManager.initialize();
// register the stores
initializeConsumer(1, 1, t1);
stateManager.register(store1, stateRestoreCallback);
@@ -266,8 +290,7 @@ public class GlobalStateManagerImplTest {
@Test(expected = ProcessorStateException.class)
public void shouldThrowProcessorStateStoreExceptionIfStoreFlushFailed() {
- stateManager.initialize(context);
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+ stateManager.initialize();
// register the stores
initializeConsumer(1, 1, t1);
stateManager.register(new NoOpReadOnlyStore(store1.name()) {
@@ -282,8 +305,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldCloseStateStores() throws IOException {
- stateManager.initialize(context);
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+ stateManager.initialize();
// register the stores
initializeConsumer(1, 1, t1);
stateManager.register(store1, stateRestoreCallback);
@@ -297,8 +319,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldWriteCheckpointsOnClose() throws IOException {
- stateManager.initialize(context);
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+ stateManager.initialize();
initializeConsumer(1, 1, t1);
stateManager.register(store1, stateRestoreCallback);
final Map<TopicPartition, Long> expected = Collections.singletonMap(t1, 25L);
@@ -309,7 +330,7 @@ public class GlobalStateManagerImplTest {
@Test(expected = ProcessorStateException.class)
public void shouldThrowProcessorStateStoreExceptionIfStoreCloseFailed() throws IOException {
- stateManager.initialize(context);
+ stateManager.initialize();
initializeConsumer(1, 1, t1);
stateManager.register(new NoOpReadOnlyStore(store1.name()) {
@Override
@@ -323,7 +344,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldThrowIllegalArgumentExceptionIfCallbackIsNull() {
- stateManager.initialize(context);
+ stateManager.initialize();
try {
stateManager.register(store1, null);
fail("should have thrown due to null callback");
@@ -334,9 +355,9 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldUnlockGlobalStateDirectoryOnClose() throws IOException {
- stateManager.initialize(context);
+ stateManager.initialize();
stateManager.close(Collections.<TopicPartition, Long>emptyMap());
- final StateDirectory stateDir = new StateDirectory(config, new MockTime());
+ final StateDirectory stateDir = new StateDirectory(streamsConfig, new MockTime());
try {
// should be able to get the lock now as it should've been released in close
assertTrue(stateDir.lockGlobalState());
@@ -347,7 +368,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldNotCloseStoresIfCloseAlreadyCalled() throws IOException {
- stateManager.initialize(context);
+ stateManager.initialize();
initializeConsumer(1, 1, t1);
stateManager.register(new NoOpReadOnlyStore("t1-store") {
@Override
@@ -366,7 +387,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldAttemptToCloseAllStoresEvenWhenSomeException() throws IOException {
- stateManager.initialize(context);
+ stateManager.initialize();
initializeConsumer(1, 1, t1);
initializeConsumer(1, 1, t2);
final NoOpReadOnlyStore store = new NoOpReadOnlyStore("t1-store") {
@@ -393,11 +414,11 @@ public class GlobalStateManagerImplTest {
public void shouldReleaseLockIfExceptionWhenLoadingCheckpoints() throws IOException {
writeCorruptCheckpoint();
try {
- stateManager.initialize(context);
+ stateManager.initialize();
} catch (StreamsException e) {
// expected
}
- final StateDirectory stateDir = new StateDirectory(config, new MockTime());
+ final StateDirectory stateDir = new StateDirectory(streamsConfig, new MockTime());
try {
// should be able to get the lock now as it should've been released
assertTrue(stateDir.lockGlobalState());
@@ -409,7 +430,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldCheckpointOffsets() throws IOException {
final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
- stateManager.initialize(context);
+ stateManager.initialize();
stateManager.checkpoint(offsets);
@@ -420,8 +441,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldNotRemoveOffsetsOfUnUpdatedTablesDuringCheckpoint() {
- stateManager.initialize(context);
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+ stateManager.initialize();
initializeConsumer(10, 1, t1);
stateManager.register(store1, stateRestoreCallback);
initializeConsumer(20, 1, t2);
@@ -450,8 +470,7 @@ public class GlobalStateManagerImplTest {
final byte[] expectedValue = "value".getBytes();
consumer.addRecord(new ConsumerRecord<>(t1.topic(), t1.partition(), 2, expectedKey, expectedValue));
- stateManager.initialize(context);
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+ stateManager.initialize();
stateManager.register(store1, stateRestoreCallback);
final KeyValue<byte[], byte[]> restoredKv = stateRestoreCallback.restored.get(0);
assertThat(stateRestoreCallback.restored, equalTo(Collections.singletonList(KeyValue.pair(restoredKv.key, restoredKv.value))));
@@ -459,8 +478,7 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldCheckpointRestoredOffsetsToFile() throws IOException {
- stateManager.initialize(context);
- final TheStateRestoreCallback stateRestoreCallback = new TheStateRestoreCallback();
+ stateManager.initialize();
initializeConsumer(10, 1, t1);
stateManager.register(store1, stateRestoreCallback);
stateManager.close(Collections.<TopicPartition, Long>emptyMap());
@@ -478,15 +496,15 @@ public class GlobalStateManagerImplTest {
@Test
public void shouldThrowLockExceptionIfIOExceptionCaughtWhenTryingToLockStateDir() {
- stateManager = new GlobalStateManagerImpl(new LogContext("mock"), topology, consumer, new StateDirectory(config, time) {
+ stateManager = new GlobalStateManagerImpl(new LogContext("mock"), topology, consumer, new StateDirectory(streamsConfig, time) {
@Override
public boolean lockGlobalState() throws IOException {
throw new IOException("KABOOM!");
}
- }, stateRestoreListener, config);
+ }, stateRestoreListener, streamsConfig);
try {
- stateManager.initialize(context);
+ stateManager.initialize();
fail("Should have thrown LockException");
} catch (final LockException e) {
// pass
@@ -504,7 +522,7 @@ public class GlobalStateManagerImplTest {
throw new TimeoutException();
}
};
- config = new StreamsConfig(new Properties() {
+ streamsConfig = new StreamsConfig(new Properties() {
{
put(StreamsConfig.APPLICATION_ID_CONFIG, "appId");
put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
@@ -520,7 +538,7 @@ public class GlobalStateManagerImplTest {
consumer,
stateDirectory,
stateRestoreListener,
- config);
+ streamsConfig);
} catch (final StreamsException expected) {
assertEquals(numberOfCalls.get(), retries);
}
@@ -537,7 +555,7 @@ public class GlobalStateManagerImplTest {
throw new TimeoutException();
}
};
- config = new StreamsConfig(new Properties() {
+ streamsConfig = new StreamsConfig(new Properties() {
{
put(StreamsConfig.APPLICATION_ID_CONFIG, "appId");
put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
@@ -553,12 +571,74 @@ public class GlobalStateManagerImplTest {
consumer,
stateDirectory,
stateRestoreListener,
- config);
+ streamsConfig);
} catch (final StreamsException expected) {
assertEquals(numberOfCalls.get(), retries);
}
}
+ @Test
+ public void shouldDeleteAndRecreateStoreDirectoryOnReinitialize() throws IOException {
+ final File storeDirectory1 = new File(stateDirectory.globalStateDir().getAbsolutePath()
+ + File.separator + "rocksdb"
+ + File.separator + storeName1);
+ final File storeDirectory2 = new File(stateDirectory.globalStateDir().getAbsolutePath()
+ + File.separator + "rocksdb"
+ + File.separator + storeName2);
+ final File storeDirectory3 = new File(stateDirectory.globalStateDir().getAbsolutePath()
+ + File.separator + storeName3);
+ final File storeDirectory4 = new File(stateDirectory.globalStateDir().getAbsolutePath()
+ + File.separator + storeName4);
+ final File testFile1 = new File(storeDirectory1.getAbsolutePath() + File.separator + "testFile");
+ final File testFile2 = new File(storeDirectory2.getAbsolutePath() + File.separator + "testFile");
+ final File testFile3 = new File(storeDirectory3.getAbsolutePath() + File.separator + "testFile");
+ final File testFile4 = new File(storeDirectory4.getAbsolutePath() + File.separator + "testFile");
+
+ consumer.updatePartitions(t1.topic(), Collections.singletonList(new PartitionInfo(t1.topic(), t1.partition(), null, null, null)));
+ consumer.updatePartitions(t2.topic(), Collections.singletonList(new PartitionInfo(t2.topic(), t2.partition(), null, null, null)));
+ consumer.updatePartitions(t3.topic(), Collections.singletonList(new PartitionInfo(t3.topic(), t3.partition(), null, null, null)));
+ consumer.updatePartitions(t4.topic(), Collections.singletonList(new PartitionInfo(t4.topic(), t4.partition(), null, null, null)));
+ consumer.updateBeginningOffsets(new HashMap<TopicPartition, Long>() {
+ {
+ put(t1, 0L);
+ put(t2, 0L);
+ put(t3, 0L);
+ put(t4, 0L);
+ }
+ });
+ consumer.updateEndOffsets(new HashMap<TopicPartition, Long>() {
+ {
+ put(t1, 0L);
+ put(t2, 0L);
+ put(t3, 0L);
+ put(t4, 0L);
+ }
+ });
+
+ stateManager.initialize();
+ stateManager.register(store1, stateRestoreCallback);
+ stateManager.register(store2, stateRestoreCallback);
+ stateManager.register(store3, stateRestoreCallback);
+ stateManager.register(store4, stateRestoreCallback);
+
+ testFile1.createNewFile();
+ assertTrue(testFile1.exists());
+ testFile2.createNewFile();
+ assertTrue(testFile2.exists());
+ testFile3.createNewFile();
+ assertTrue(testFile3.exists());
+ testFile4.createNewFile();
+ assertTrue(testFile4.exists());
+
+ // only delete and recreate store 1 and 3 -- 2 and 4 must be untouched
+ stateManager.reinitializeStateStoresForPartitions(Utils.mkList(t1, t3), mockProcessorContext);
+
+ assertFalse(testFile1.exists());
+ assertTrue(testFile2.exists());
+ assertFalse(testFile3.exists());
+ assertTrue(testFile4.exists());
+ }
+
private void writeCorruptCheckpoint() throws IOException {
final File checkpointFile = new File(stateManager.baseDir(), ProcessorStateManager.CHECKPOINT_FILE_NAME);
try (final FileOutputStream stream = new FileOutputStream(checkpointFile)) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
index 2bd2d42..c71f469 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
@@ -16,17 +16,26 @@
*/
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.StreamsException;
-import org.apache.kafka.streams.kstream.KStreamBuilder;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.kstream.internals.InternalNameProvider;
+import org.apache.kafka.streams.kstream.internals.KTableSource;
+import org.apache.kafka.streams.kstream.internals.KeyValueStoreMaterializer;
+import org.apache.kafka.streams.kstream.internals.MaterializedInternal;
import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.test.MockStateRestoreListener;
import org.apache.kafka.test.TestCondition;
import org.apache.kafka.test.TestUtils;
@@ -36,7 +45,9 @@ import org.junit.Test;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Set;
+import static org.apache.kafka.streams.processor.internals.GlobalStreamThread.State.DEAD;
import static org.apache.kafka.streams.processor.internals.GlobalStreamThread.State.RUNNING;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -47,20 +58,49 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class GlobalStreamThreadTest {
- private final KStreamBuilder builder = new KStreamBuilder();
- private final MockConsumer<byte[], byte[]> mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+ private final InternalTopologyBuilder builder = new InternalTopologyBuilder();
+ private final MockConsumer<byte[], byte[]> mockConsumer = new MockConsumer<>(OffsetResetStrategy.NONE);
private final MockTime time = new MockTime();
private final MockStateRestoreListener stateRestoreListener = new MockStateRestoreListener();
private GlobalStreamThread globalStreamThread;
private StreamsConfig config;
+ private final static String GLOBAL_STORE_TOPIC_NAME = "foo";
+ private final static String GLOBAL_STORE_NAME = "bar";
+ private final TopicPartition topicPartition = new TopicPartition(GLOBAL_STORE_TOPIC_NAME, 0);
+
+ @SuppressWarnings("unchecked")
@Before
public void before() {
- builder.globalTable("foo", "bar");
+ final MaterializedInternal<Object, Object, KeyValueStore<Bytes, byte[]>> materialized = new MaterializedInternal<>(
+ Materialized.<Object, Object, KeyValueStore<Bytes, byte[]>>with(null, null),
+ new InternalNameProvider() {
+ @Override
+ public String newProcessorName(String prefix) {
+ return "processorName";
+ }
+
+ @Override
+ public String newStoreName(String prefix) {
+ return GLOBAL_STORE_NAME;
+ }
+ },
+ "store-");
+
+ builder.addGlobalStore(
+ (StoreBuilder) new KeyValueStoreMaterializer<>(materialized).materialize().withLoggingDisabled(),
+ "sourceName",
+ null,
+ null,
+ null,
+ GLOBAL_STORE_TOPIC_NAME,
+ "processorName",
+ new KTableSource<>(GLOBAL_STORE_NAME));
+
final HashMap<String, Object> properties = new HashMap<>();
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "blah");
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "blah");
- properties.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
+ properties.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
config = new StreamsConfig(properties);
globalStreamThread = new GlobalStreamThread(builder.buildGlobalStateTopology(),
config,
@@ -115,7 +155,6 @@ public class GlobalStreamThreadTest {
assertFalse(globalStreamThread.stillRunning());
}
-
@Test
public void shouldBeRunningAfterSuccessfulStart() {
initializeConsumer();
@@ -136,7 +175,7 @@ public class GlobalStreamThreadTest {
public void shouldCloseStateStoresOnClose() throws InterruptedException {
initializeConsumer();
globalStreamThread.start();
- final StateStore globalStore = builder.globalStateStores().get("bar");
+ final StateStore globalStore = builder.globalStateStores().get(GLOBAL_STORE_NAME);
assertTrue(globalStore.isOpen());
globalStreamThread.shutdown();
globalStreamThread.join();
@@ -146,7 +185,6 @@ public class GlobalStreamThreadTest {
@SuppressWarnings("unchecked")
@Test
public void shouldTransitionToDeadOnClose() throws InterruptedException {
-
initializeConsumer();
globalStreamThread.start();
globalStreamThread.shutdown();
@@ -158,7 +196,6 @@ public class GlobalStreamThreadTest {
@SuppressWarnings("unchecked")
@Test
public void shouldStayDeadAfterTwoCloses() throws InterruptedException {
-
initializeConsumer();
globalStreamThread.start();
globalStreamThread.shutdown();
@@ -170,8 +207,7 @@ public class GlobalStreamThreadTest {
@SuppressWarnings("unchecked")
@Test
- public void shouldTransitiontoRunningOnStart() throws InterruptedException {
-
+ public void shouldTransitionToRunningOnStart() throws InterruptedException {
initializeConsumer();
globalStreamThread.start();
TestUtils.waitForCondition(new TestCondition() {
@@ -183,19 +219,52 @@ public class GlobalStreamThreadTest {
globalStreamThread.shutdown();
}
+ @Test
+ public void shouldDieOnInvalidOffsetException() throws Exception {
+ initializeConsumer();
+ globalStreamThread.start();
+ TestUtils.waitForCondition(new TestCondition() {
+ @Override
+ public boolean conditionMet() {
+ return globalStreamThread.state() == RUNNING;
+ }
+ }, 10 * 1000, "Thread never started.");
+ mockConsumer.updateEndOffsets(Collections.singletonMap(topicPartition, 1L));
+ mockConsumer.addRecord(new ConsumerRecord<>(GLOBAL_STORE_TOPIC_NAME, 0, 0L, "K1".getBytes(), "V1".getBytes()));
+
+ TestUtils.waitForCondition(new TestCondition() {
+ @Override
+ public boolean conditionMet() {
+ return mockConsumer.position(topicPartition) == 1L;
+ }
+ }, 10 * 1000, "Input record never consumed");
+
+ mockConsumer.setException(new InvalidOffsetException("Try Again!") {
+ @Override
+ public Set<TopicPartition> partitions() {
+ return Collections.singleton(topicPartition);
+ }
+ });
+ // feed first record for recovery
+ mockConsumer.addRecord(new ConsumerRecord<>(GLOBAL_STORE_TOPIC_NAME, 0, 0L, "K1".getBytes(), "V1".getBytes()));
+
+ TestUtils.waitForCondition(new TestCondition() {
+ @Override
+ public boolean conditionMet() {
+ return globalStreamThread.state() == DEAD;
+ }
+ }, 10 * 1000, "GlobalStreamThread should have died.");
+ }
private void initializeConsumer() {
- mockConsumer.updatePartitions("foo", Collections.singletonList(new PartitionInfo("foo",
- 0,
- null,
- new Node[0],
- new Node[0])));
- final TopicPartition topicPartition = new TopicPartition("foo", 0);
+ mockConsumer.updatePartitions(GLOBAL_STORE_TOPIC_NAME, Collections.singletonList(new PartitionInfo(GLOBAL_STORE_TOPIC_NAME,
+ 0,
+ null,
+ new Node[0],
+ new Node[0])));
mockConsumer.updateBeginningOffsets(Collections.singletonMap(topicPartition, 0L));
mockConsumer.updateEndOffsets(Collections.singletonMap(topicPartition, 0L));
+ mockConsumer.assign(Collections.singleton(topicPartition));
}
-
-
-
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
index 2bb5b7b..725211d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
@@ -135,7 +135,7 @@ public class StateConsumerTest {
private boolean flushed;
private boolean closed;
- public StateMaintainerStub(final Map<TopicPartition, Long> partitionOffsets) {
+ StateMaintainerStub(final Map<TopicPartition, Long> partitionOffsets) {
this.partitionOffsets = partitionOffsets;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerStub.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerStub.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerStub.java
index 47a0015..b14731d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerStub.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerStub.java
@@ -23,6 +23,7 @@ import org.apache.kafka.streams.processor.StateStore;
import java.io.File;
import java.io.IOException;
+import java.util.Collection;
import java.util.Map;
public class StateManagerStub implements StateManager {
@@ -33,7 +34,12 @@ public class StateManagerStub implements StateManager {
}
@Override
- public void register(final StateStore store, final StateRestoreCallback stateRestoreCallback) {}
+ public void register(final StateStore store,
+ final StateRestoreCallback stateRestoreCallback) {}
+
+ @Override
+ public void reinitializeStateStoresForPartitions(final Collection<TopicPartition> partitions,
+ final InternalProcessorContext processorContext) {}
@Override
public void flush() {}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java
----------------------------------------------------------------------
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 342354c..9f6f712 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
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.PartitionInfo;
@@ -43,6 +44,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.kafka.test.MockStateRestoreListener.RESTORE_BATCH;
@@ -52,6 +54,7 @@ import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.replay;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.IsEqual.equalTo;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -124,6 +127,30 @@ public class StoreChangelogReaderTest {
}
@Test
+ public void shouldRecoverFromInvalidOffsetExceptionAndFinishRestore() {
+ final int messages = 10;
+ setupConsumer(messages, topicPartition);
+ consumer.setException(new InvalidOffsetException("Try Again!") {
+ @Override
+ public Set<TopicPartition> partitions() {
+ return Collections.singleton(topicPartition);
+ }
+ });
+ changelogReader.register(new StateRestorer(topicPartition, restoreListener, null, Long.MAX_VALUE, true,
+ "storeName"));
+
+ EasyMock.expect(active.restoringTaskFor(topicPartition)).andReturn(task);
+ EasyMock.replay(active);
+
+ // first restore call "fails" but we should not die with an exception
+ assertEquals(0, changelogReader.restore(active).size());
+ // retry restore should succeed
+ assertEquals(1, changelogReader.restore(active).size());
+ assertThat(callback.restored.size(), equalTo(messages));
+ }
+
+
+ @Test
public void shouldRestoreMessagesFromCheckpoint() {
final int messages = 10;
setupConsumer(messages, topicPartition);
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index c852ae3..4aee8f5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -35,9 +35,6 @@ import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.errors.StreamsException;
-import org.apache.kafka.streams.processor.AbstractProcessor;
-import org.apache.kafka.streams.processor.Processor;
-import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.processor.StateRestoreListener;
@@ -49,7 +46,6 @@ import org.apache.kafka.test.MockSourceNode;
import org.apache.kafka.test.MockStateRestoreListener;
import org.apache.kafka.test.MockStateStore;
import org.apache.kafka.test.MockTimestampExtractor;
-import org.apache.kafka.test.NoOpProcessorContext;
import org.apache.kafka.test.NoOpRecordCollector;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
@@ -465,12 +461,12 @@ public class StreamTaskTest {
task.process();
fail("Should've thrown StreamsException");
} catch (final Exception e) {
- assertThat(((ProcessorContextImpl) task.processorContext()).currentNode(), nullValue());
+ assertThat(task.processorContext.currentNode(), nullValue());
}
}
@Test
- public void shouldWrapKafkaExceptionsWithStreamsExceptionAndAddContextWhenPunctuating() {
+ public void shouldWrapKafkaExceptionsWithStreamsExceptionAndAddContextWhenPunctuatingStreamTime() {
task = createStatelessTask(false);
task.initialize();
@@ -483,32 +479,19 @@ public class StreamTaskTest {
});
fail("Should've thrown StreamsException");
} catch (final StreamsException e) {
- assertThat(((ProcessorContextImpl) task.processorContext()).currentNode(), nullValue());
+ final String message = e.getMessage();
+ assertTrue("message=" + message + " should contain processor", message.contains("processor '" + processorStreamTime.name() + "'"));
+ assertThat(task.processorContext.currentNode(), nullValue());
}
}
@Test
- public void shouldWrapKafkaExceptionsWithStreamsExceptionAndAddContextWhenPunctuatingStreamTime() {
- final Processor<Object, Object> processor = new AbstractProcessor<Object, Object>() {
- @Override
- public void init(final ProcessorContext context) {
- }
-
- @Override
- public void process(final Object key, final Object value) {}
-
- @Override
- public void punctuate(final long timestamp) {}
- };
-
- final ProcessorNode<Object, Object> punctuator = new ProcessorNode<>("test", processor, Collections.<String>emptySet());
- punctuator.init(new NoOpProcessorContext());
-
+ public void shouldWrapKafkaExceptionsWithStreamsExceptionAndAddContextWhenPunctuatingWallClockTimeTime() {
task = createStatelessTask(false);
task.initialize();
try {
- task.punctuate(punctuator, 1, PunctuationType.STREAM_TIME, new Punctuator() {
+ task.punctuate(processorSystemTime, 1, PunctuationType.WALL_CLOCK_TIME, new Punctuator() {
@Override
public void punctuate(long timestamp) {
throw new KafkaException("KABOOM!");
@@ -516,7 +499,9 @@ public class StreamTaskTest {
});
fail("Should've thrown StreamsException");
} catch (final StreamsException e) {
- assertThat(((ProcessorContextImpl) task.processorContext()).currentNode(), nullValue());
+ final String message = e.getMessage();
+ assertTrue("message=" + message + " should contain processor", message.contains("processor '" + processorSystemTime.name() + "'"));
+ assertThat(task.processorContext.currentNode(), nullValue());
}
}
@@ -567,7 +552,7 @@ public class StreamTaskTest {
public void shouldThrowIllegalStateExceptionIfCurrentNodeIsNotNullWhenPunctuateCalled() {
task = createStatelessTask(false);
task.initialize();
- ((ProcessorContextImpl) task.processorContext()).setCurrentNode(processorStreamTime);
+ task.processorContext.setCurrentNode(processorStreamTime);
try {
task.punctuate(processorStreamTime, 10, PunctuationType.STREAM_TIME, punctuator);
fail("Should throw illegal state exception as current node is not null");
@@ -591,7 +576,7 @@ public class StreamTaskTest {
task = createStatelessTask(false);
task.initialize();
task.punctuate(processorStreamTime, 5, PunctuationType.STREAM_TIME, punctuator);
- assertThat(((ProcessorContextImpl) task.processorContext()).currentNode(), nullValue());
+ assertThat(((ProcessorContextImpl) task.context()).currentNode(), nullValue());
}
@Test(expected = IllegalStateException.class)
@@ -608,7 +593,7 @@ public class StreamTaskTest {
@Test
public void shouldNotThrowExceptionOnScheduleIfCurrentNodeIsNotNull() {
task = createStatelessTask(false);
- ((ProcessorContextImpl) task.processorContext()).setCurrentNode(processorStreamTime);
+ task.processorContext.setCurrentNode(processorStreamTime);
task.schedule(1, PunctuationType.STREAM_TIME, new Punctuator() {
@Override
public void punctuate(long timestamp) {
@@ -744,7 +729,7 @@ public class StreamTaskTest {
}
@Test
- public void shouldNotAbortTransactionOnZombieClosedIfEosEnabled() throws Exception {
+ public void shouldNotAbortTransactionOnZombieClosedIfEosEnabled() {
task = createStatelessTask(true);
task.close(false, true);
task = null;
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index 69ae07c..8bcd6fb 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
@@ -26,17 +27,20 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.TaskMigratedException;
+import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.internals.ConsumedInternal;
import org.apache.kafka.streams.kstream.internals.InternalStreamsBuilder;
import org.apache.kafka.streams.kstream.internals.InternalStreamsBuilderTest;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.TaskMetadata;
import org.apache.kafka.streams.processor.ThreadMetadata;
+import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.test.MockClientSupplier;
import org.apache.kafka.test.MockStateRestoreListener;
import org.apache.kafka.test.MockTimestampExtractor;
@@ -631,7 +635,8 @@ public class StreamThreadTest {
@Test
public void shouldReturnStandbyTaskMetadataWhileRunningState() throws InterruptedException {
- internalStreamsBuilder.stream(Collections.singleton(topic1), consumed).groupByKey().count("count-one");
+ internalStreamsBuilder.stream(Collections.singleton(topic1), consumed)
+ .groupByKey().count(Materialized.<Object, Long, KeyValueStore<Bytes, byte[]>>as("count-one"));
final StreamThread thread = createStreamThread(clientId, config, false);
final MockConsumer<byte[], byte[]> restoreConsumer = clientSupplier.restoreConsumer;
@@ -681,7 +686,8 @@ public class StreamThreadTest {
@Test
public void shouldAlwaysReturnEmptyTasksMetadataWhileRebalancingStateAndTasksNotRunning() throws InterruptedException {
- internalStreamsBuilder.stream(Collections.singleton(topic1), consumed).groupByKey().count("count-one");
+ internalStreamsBuilder.stream(Collections.singleton(topic1), consumed)
+ .groupByKey().count(Materialized.<Object, Long, KeyValueStore<Bytes, byte[]>>as("count-one"));
final StreamThread thread = createStreamThread(clientId, config, false);
final MockConsumer<byte[], byte[]> restoreConsumer = clientSupplier.restoreConsumer;
@@ -728,6 +734,97 @@ public class StreamThreadTest {
assertThreadMetadataHasEmptyTasksWithState(thread.threadMetadata(), StreamThread.State.PARTITIONS_ASSIGNED);
}
+ @Test
+ public void shouldRecoverFromInvalidOffsetExceptionOnRestoreAndFinishRestore() throws Exception {
+ internalStreamsBuilder.stream(Collections.singleton("topic"), consumed)
+ .groupByKey().count(Materialized.<Object, Long, KeyValueStore<Bytes, byte[]>>as("count"));
+
+ final StreamThread thread = createStreamThread("cliendId", config, false);
+ final MockConsumer<byte[], byte[]> mockConsumer = (MockConsumer<byte[], byte[]>) thread.consumer;
+ final MockConsumer<byte[], byte[]> mockRestoreConsumer = (MockConsumer<byte[], byte[]>) thread.restoreConsumer;
+
+ final TopicPartition topicPartition = new TopicPartition("topic", 0);
+ final Set<TopicPartition> topicPartitionSet = Collections.singleton(topicPartition);
+
+ final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
+ activeTasks.put(new TaskId(0, 0), topicPartitionSet);
+ thread.taskManager().setAssignmentMetadata(activeTasks, Collections.<TaskId, Set<TopicPartition>>emptyMap());
+
+ mockConsumer.updatePartitions("topic", new ArrayList<PartitionInfo>() {
+ {
+ add(new PartitionInfo("topic",
+ 0,
+ null,
+ new Node[0],
+ new Node[0]));
+ }
+ });
+ mockConsumer.updateBeginningOffsets(Collections.singletonMap(topicPartition, 0L));
+
+ mockRestoreConsumer.updatePartitions("stream-thread-test-count-changelog", new ArrayList<PartitionInfo>() {
+ {
+ add(new PartitionInfo("stream-thread-test-count-changelog",
+ 0,
+ null,
+ new Node[0],
+ new Node[0]));
+ }
+ });
+ final TopicPartition changelogPartition = new TopicPartition("stream-thread-test-count-changelog", 0);
+ final Set<TopicPartition> changelogPartitionSet = Collections.singleton(changelogPartition);
+ mockRestoreConsumer.updateBeginningOffsets(Collections.singletonMap(changelogPartition, 0L));
+ mockRestoreConsumer.updateEndOffsets(Collections.singletonMap(changelogPartition, 2L));
+
+ mockConsumer.schedulePollTask(new Runnable() {
+ @Override
+ public void run() {
+ thread.setState(StreamThread.State.PARTITIONS_REVOKED);
+ thread.rebalanceListener.onPartitionsAssigned(topicPartitionSet);
+ }
+ });
+
+ try {
+ thread.start();
+
+ TestUtils.waitForCondition(new TestCondition() {
+ @Override
+ public boolean conditionMet() {
+ return mockRestoreConsumer.assignment().size() == 1;
+ }
+ }, "Never restore first record");
+
+ mockRestoreConsumer.addRecord(new ConsumerRecord<>("stream-thread-test-count-changelog", 0, 0L, "K1".getBytes(), "V1".getBytes()));
+
+ TestUtils.waitForCondition(new TestCondition() {
+ @Override
+ public boolean conditionMet() {
+ return mockRestoreConsumer.position(changelogPartition) == 1L;
+ }
+ }, "Never restore first record");
+
+ mockRestoreConsumer.setException(new InvalidOffsetException("Try Again!") {
+ @Override
+ public Set<TopicPartition> partitions() {
+ return changelogPartitionSet;
+ }
+ });
+
+ mockRestoreConsumer.addRecord(new ConsumerRecord<>("stream-thread-test-count-changelog", 0, 0L, "K1".getBytes(), "V1".getBytes()));
+ mockRestoreConsumer.addRecord(new ConsumerRecord<>("stream-thread-test-count-changelog", 0, 1L, "K2".getBytes(), "V2".getBytes()));
+
+ TestUtils.waitForCondition(new TestCondition() {
+ @Override
+ public boolean conditionMet() {
+ mockRestoreConsumer.assign(changelogPartitionSet);
+ return mockRestoreConsumer.position(changelogPartition) == 2L;
+ }
+ }, "Never finished restore");
+ } finally {
+ thread.shutdown();
+ thread.join(10000);
+ }
+ }
+
private void assertThreadMetadataHasEmptyTasksWithState(ThreadMetadata metadata, StreamThread.State state) {
assertEquals(state.name(), metadata.threadState());
assertTrue(metadata.activeTasks().isEmpty());
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java b/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java
index 6677084..ae0b923 100644
--- a/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java
+++ b/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java
@@ -24,6 +24,7 @@ import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import java.io.File;
import java.io.IOException;
+import java.util.Collection;
import java.util.Map;
import java.util.Set;
@@ -40,20 +41,26 @@ public class GlobalStateManagerStub implements GlobalStateManager {
}
@Override
- public Set<String> initialize(final InternalProcessorContext processorContext) {
+ public void setGlobalProcessorContext(InternalProcessorContext processorContext) {}
+
+ @Override
+ public Set<String> initialize() {
initialized = true;
return storeNames;
}
-
+
+ @Override
+ public void reinitializeStateStoresForPartitions(final Collection<TopicPartition> partitions,
+ final InternalProcessorContext processorContext) {}
+
@Override
public File baseDir() {
return null;
}
@Override
- public void register(final StateStore store, final StateRestoreCallback stateRestoreCallback) {
-
- }
+ public void register(final StateStore store,
+ final StateRestoreCallback stateRestoreCallback) {}
@Override
public void flush() {}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/test/NoOpReadOnlyStore.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/NoOpReadOnlyStore.java b/streams/src/test/java/org/apache/kafka/test/NoOpReadOnlyStore.java
index 0ada2e4..ae46b8d 100644
--- a/streams/src/test/java/org/apache/kafka/test/NoOpReadOnlyStore.java
+++ b/streams/src/test/java/org/apache/kafka/test/NoOpReadOnlyStore.java
@@ -21,21 +21,30 @@ import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
+import java.io.File;
+
public class NoOpReadOnlyStore<K, V>
implements ReadOnlyKeyValueStore<K, V>, StateStore {
private final String name;
+ private final boolean rocksdbStore;
private boolean open = true;
public boolean initialized;
public boolean flushed;
public NoOpReadOnlyStore() {
- this("");
+ this("", false);
}
public NoOpReadOnlyStore(final String name) {
+ this(name, false);
+ }
+
+ public NoOpReadOnlyStore(final String name,
+ final boolean rocksdbStore) {
this.name = name;
+ this.rocksdbStore = rocksdbStore;
}
@Override
@@ -65,6 +74,12 @@ public class NoOpReadOnlyStore<K, V>
@Override
public void init(final ProcessorContext context, final StateStore root) {
+ if (rocksdbStore) {
+ // cf. RocksDBStore
+ new File(context.stateDir() + File.separator + "rocksdb" + File.separator + name).mkdirs();
+ } else {
+ new File(context.stateDir() + File.separator + name).mkdir();
+ }
this.initialized = true;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java b/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
index 887b10d..ae0cc9c 100644
--- a/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
@@ -37,7 +37,6 @@ import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler;
-import org.apache.kafka.streams.processor.StateRestoreListener;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.GlobalProcessorContextImpl;
@@ -155,11 +154,9 @@ public class ProcessorTopologyTestDriver {
private final Map<String, Queue<ProducerRecord<byte[], byte[]>>> outputRecordsByTopic = new HashMap<>();
private final Set<String> internalTopics = new HashSet<>();
private final Map<String, TopicPartition> globalPartitionsByTopic = new HashMap<>();
- private final StateRestoreListener stateRestoreListener = new MockStateRestoreListener();
private StreamTask task;
private GlobalStateUpdateTask globalStateTask;
-
/**
* Create a new test diver instance
* @param config the stream configuration for the topology
@@ -227,8 +224,10 @@ public class ProcessorTopologyTestDriver {
stateDirectory,
stateRestoreListener,
config);
+ final GlobalProcessorContextImpl globalProcessorContext = new GlobalProcessorContextImpl(config, stateManager, streamsMetrics, cache);
+ stateManager.setGlobalProcessorContext(globalProcessorContext);
globalStateTask = new GlobalStateUpdateTask(globalTopology,
- new GlobalProcessorContextImpl(config, stateManager, streamsMetrics, cache),
+ globalProcessorContext,
stateManager,
new LogAndContinueExceptionHandler(),
new LogContext());
@@ -242,8 +241,8 @@ public class ProcessorTopologyTestDriver {
consumer,
new StoreChangelogReader(
createRestoreConsumer(topology.storeToChangelogTopic()),
- stateRestoreListener,
- new LogContext("topology-test-driver ")),
+ new MockStateRestoreListener(),
+ new LogContext("topology-test-driver ")),
config,
streamsMetrics, stateDirectory,
cache,
[2/2] kafka git commit: KAFKA-6121: Restore and global consumer
should not use auto.offset.reset
Posted by da...@apache.org.
KAFKA-6121: Restore and global consumer should not use auto.offset.reset
- set auto.offset.reste to "none" for restore and global consumer
- handle InvalidOffsetException for restore and global consumer
- add corresponding tests
- some minor cleanup
Author: Matthias J. Sax <ma...@confluent.io>
Reviewers: Damian Guy <damian.guy@gmail.com, Bill Bejeck <bi...@confluent.io>, GuozhangWang <wa...@gmail.com>
Closes #4215 from mjsax/kafka-6121-restore-global-consumer-handle-reset
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/04395175
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/04395175
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/04395175
Branch: refs/heads/trunk
Commit: 043951753b6fb6c8bae6d25d7a6a97e74b614cac
Parents: 2bf2348
Author: Matthias J. Sax <ma...@confluent.io>
Authored: Mon Dec 11 14:20:10 2017 +0000
Committer: Damian Guy <da...@gmail.com>
Committed: Mon Dec 11 14:20:10 2017 +0000
----------------------------------------------------------------------
.../kafka/clients/consumer/MockConsumer.java | 8 +
.../org/apache/kafka/streams/StreamsConfig.java | 1 +
.../internals/AbstractProcessorContext.java | 5 +
.../internals/AbstractStateManager.java | 114 ++++++++++
.../processor/internals/AbstractTask.java | 6 +-
.../processor/internals/GlobalStateManager.java | 5 +-
.../internals/GlobalStateManagerImpl.java | 90 +++++---
.../internals/GlobalStateUpdateTask.java | 4 +-
.../processor/internals/GlobalStreamThread.java | 38 ++--
.../internals/InternalProcessorContext.java | 7 +-
.../internals/ProcessorStateManager.java | 70 +++---
.../processor/internals/StandbyTask.java | 6 +-
.../processor/internals/StateManager.java | 6 +-
.../internals/StoreChangelogReader.java | 21 +-
.../streams/processor/internals/StreamTask.java | 6 -
.../processor/internals/StreamThread.java | 33 ++-
.../internals/InnerMeteredKeyValueStore.java | 1 -
.../internals/MeteredKeyValueBytesStore.java | 2 -
.../apache/kafka/streams/StreamsConfigTest.java | 6 -
.../processor/internals/AbstractTaskTest.java | 147 +++++++++++--
.../internals/GlobalStateManagerImplTest.java | 216 +++++++++++++------
.../internals/GlobalStreamThreadTest.java | 111 ++++++++--
.../processor/internals/StateConsumerTest.java | 2 +-
.../processor/internals/StateManagerStub.java | 8 +-
.../internals/StoreChangelogReaderTest.java | 27 +++
.../processor/internals/StreamTaskTest.java | 43 ++--
.../processor/internals/StreamThreadTest.java | 101 ++++++++-
.../kafka/test/GlobalStateManagerStub.java | 17 +-
.../apache/kafka/test/NoOpReadOnlyStore.java | 17 +-
.../kafka/test/ProcessorTopologyTestDriver.java | 11 +-
30 files changed, 862 insertions(+), 267 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
----------------------------------------------------------------------
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 9b0c058..10aedbb 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
@@ -108,6 +108,14 @@ public class MockConsumer<K, V> implements Consumer<K, V> {
}
ensureNotClosed();
this.subscriptions.subscribeFromPattern(topicsToSubscribe);
+ final Set<TopicPartition> assignedPartitions = new HashSet<>();
+ for (final String topic : topicsToSubscribe) {
+ for (final PartitionInfo info : this.partitions.get(topic)) {
+ assignedPartitions.add(new TopicPartition(topic, info.partition()));
+ }
+
+ }
+ subscriptions.assignFromSubscribed(assignedPartitions);
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index 49b8a3c..d78fc0d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -763,6 +763,7 @@ public class StreamsConfig extends AbstractConfig {
consumerProps.remove(ConsumerConfig.GROUP_ID_CONFIG);
// add client id with stream client id prefix
consumerProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-restore-consumer");
+ consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
return consumerProps;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
index aa58226..87408c6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
@@ -197,4 +197,9 @@ public abstract class AbstractProcessorContext implements InternalProcessorConte
public void initialized() {
initialized = true;
}
+
+ @Override
+ public void uninitialize() {
+ initialized = false;
+ }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java
new file mode 100644
index 0000000..777e46b
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
+import org.slf4j.Logger;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+
+abstract class AbstractStateManager implements StateManager {
+ static final String CHECKPOINT_FILE_NAME = ".checkpoint";
+
+ final File baseDir;
+ final Map<TopicPartition, Long> checkpointableOffsets = new HashMap<>();
+
+ OffsetCheckpoint checkpoint;
+
+ final Map<String, StateStore> stores = new LinkedHashMap<>();
+ final Map<String, StateStore> globalStores = new LinkedHashMap<>();
+
+ AbstractStateManager(final File baseDir) {
+ this.baseDir = baseDir;
+ this.checkpoint = new OffsetCheckpoint(new File(baseDir, CHECKPOINT_FILE_NAME));
+
+ }
+
+ public void reinitializeStateStoresForPartitions(final Logger log,
+ final Map<String, StateStore> stateStores,
+ final Map<String, String> storeToChangelogTopic,
+ final Collection<TopicPartition> partitions,
+ final InternalProcessorContext processorContext) {
+ final Map<String, String> changelogTopicToStore = inverseOneToOneMap(storeToChangelogTopic);
+ final Set<String> storeToBeReinitialized = new HashSet<>();
+
+ for (final TopicPartition topicPartition : partitions) {
+ checkpointableOffsets.remove(topicPartition);
+ storeToBeReinitialized.add(changelogTopicToStore.get(topicPartition.topic()));
+ }
+ try {
+ checkpoint.write(checkpointableOffsets);
+ } catch (final IOException fatalException) {
+ log.error("Failed to update checkpoint file for global stores.", fatalException);
+ throw new StreamsException("Failed to reinitialize global store.", fatalException);
+ }
+
+ final Iterator<Map.Entry<String, StateStore>> it = stateStores.entrySet().iterator();
+ while (it.hasNext()) {
+ final StateStore stateStore = it.next().getValue();
+ final String storeName = stateStore.name();
+ if (storeToBeReinitialized.contains(storeName)) {
+ try {
+ stateStore.close();
+ } catch (final RuntimeException ignoreAndSwallow) { /* ignore */ }
+ processorContext.uninitialize();
+ it.remove();
+
+ // TODO remove this eventually
+ // -> (only after we are sure, we don't need it for backward compatibility reasons anymore; maybe 2.0 release?)
+ // this is an ugly "hack" that is required because RocksDBStore does not follow the pattern to put the
+ // store directory as <taskDir>/<storeName> but nests it with an intermediate <taskDir>/rocksdb/<storeName>
+ try {
+ Utils.delete(new File(baseDir + File.separator + "rocksdb" + File.separator + storeName));
+ } catch (final IOException fatalException) {
+ log.error("Failed to reinitialize store {}.", storeName, fatalException);
+ throw new StreamsException(String.format("Failed to reinitialize store %s.", storeName), fatalException);
+ }
+
+ try {
+ Utils.delete(new File(baseDir + File.separator + storeName));
+ } catch (final IOException fatalException) {
+ log.error("Failed to reinitialize store {}.", storeName, fatalException);
+ throw new StreamsException(String.format("Failed to reinitialize store %s.", storeName), fatalException);
+ }
+
+ stateStore.init(processorContext, stateStore);
+ }
+ }
+ }
+
+ private Map<String, String> inverseOneToOneMap(final Map<String, String> origin) {
+ final Map<String, String> reversedMap = new HashMap<>();
+ for (final Map.Entry<String, String> entry : origin.entrySet()) {
+ reversedMap.put(entry.getValue(), entry.getKey());
+ }
+ return reversedMap;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
index d11af3a..2b8af6d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
@@ -52,7 +52,7 @@ public abstract class AbstractTask implements Task {
final Logger log;
final LogContext logContext;
boolean taskInitialized;
- private final StateDirectory stateDirectory;
+ final StateDirectory stateDirectory;
InternalProcessorContext processorContext;
@@ -220,10 +220,14 @@ public abstract class AbstractTask implements Task {
for (final StateStore store : topology.stateStores()) {
log.trace("Initializing store {}", store.name());
+ processorContext.uninitialize();
store.init(processorContext, store);
}
}
+ void reinitializeStateStoresForPartitions(final Collection<TopicPartition> partitions) {
+ stateMgr.reinitializeStateStoresForPartitions(partitions, processorContext);
+ }
/**
* @throws ProcessorStateException if there is an error while closing the state manager
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManager.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManager.java
index c9b8ca8..479fd1f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManager.java
@@ -21,9 +21,12 @@ import org.apache.kafka.streams.errors.StreamsException;
import java.util.Set;
public interface GlobalStateManager extends StateManager {
+
+ void setGlobalProcessorContext(final InternalProcessorContext processorContext);
+
/**
* @throws IllegalStateException If store gets registered after initialized is already finished
* @throws StreamsException if the store's change log does not contain the partition
*/
- Set<String> initialize(InternalProcessorContext processorContext);
+ Set<String> initialize();
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
index bbae9aa..2d4ee8f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
@@ -19,6 +19,7 @@ package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.TimeoutException;
@@ -33,38 +34,30 @@ import org.apache.kafka.streams.processor.BatchingStateRestoreCallback;
import org.apache.kafka.streams.processor.StateRestoreCallback;
import org.apache.kafka.streams.processor.StateRestoreListener;
import org.apache.kafka.streams.processor.StateStore;
-import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
import org.slf4j.Logger;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import static org.apache.kafka.streams.processor.internals.ProcessorStateManager.CHECKPOINT_FILE_NAME;
-
/**
* This class is responsible for the initialization, restoration, closing, flushing etc
* of Global State Stores. There is only ever 1 instance of this class per Application Instance.
*/
-public class GlobalStateManagerImpl implements GlobalStateManager {
+public class GlobalStateManagerImpl extends AbstractStateManager implements GlobalStateManager {
private final Logger log;
-
private final ProcessorTopology topology;
private final Consumer<byte[], byte[]> globalConsumer;
private final StateDirectory stateDirectory;
- private final Map<String, StateStore> stores = new LinkedHashMap<>();
- private final File baseDir;
- private final OffsetCheckpoint checkpoint;
private final Set<String> globalStoreNames = new HashSet<>();
- private final Map<TopicPartition, Long> checkpointableOffsets = new HashMap<>();
private final StateRestoreListener stateRestoreListener;
+ private InternalProcessorContext processorContext;
private final int retries;
private final long retryBackoffMs;
@@ -74,19 +67,24 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
final StateDirectory stateDirectory,
final StateRestoreListener stateRestoreListener,
final StreamsConfig config) {
+ super(stateDirectory.globalStateDir());
+
this.log = logContext.logger(GlobalStateManagerImpl.class);
this.topology = topology;
this.globalConsumer = globalConsumer;
this.stateDirectory = stateDirectory;
- this.baseDir = stateDirectory.globalStateDir();
- this.checkpoint = new OffsetCheckpoint(new File(this.baseDir, CHECKPOINT_FILE_NAME));
this.stateRestoreListener = stateRestoreListener;
this.retries = config.getInt(StreamsConfig.RETRIES_CONFIG);
this.retryBackoffMs = config.getLong(StreamsConfig.RETRY_BACKOFF_MS_CONFIG);
}
@Override
- public Set<String> initialize(final InternalProcessorContext processorContext) {
+ public void setGlobalProcessorContext(final InternalProcessorContext processorContext) {
+ this.processorContext = processorContext;
+ }
+
+ @Override
+ public Set<String> initialize() {
try {
if (!stateDirectory.lockGlobalState()) {
throw new LockException(String.format("Failed to lock the global state directory: %s", baseDir));
@@ -103,7 +101,7 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
} catch (IOException e1) {
log.error("Failed to unlock the global state directory", e);
}
- throw new StreamsException("Failed to read checkpoints for global state stores", e);
+ throw new StreamsException("Failed to read checkpoints for global state globalStores", e);
}
final List<StateStore> stateStores = topology.globalStateStores();
@@ -115,8 +113,22 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
}
@Override
+ public void reinitializeStateStoresForPartitions(final Collection<TopicPartition> partitions,
+ final InternalProcessorContext processorContext) {
+ super.reinitializeStateStoresForPartitions(
+ log,
+ globalStores,
+ topology.storeToChangelogTopic(),
+ partitions,
+ processorContext);
+
+ globalConsumer.assign(partitions);
+ globalConsumer.seekToBeginning(partitions);
+ }
+
+ @Override
public StateStore getGlobalStore(final String name) {
- return stores.get(name);
+ return globalStores.get(name);
}
@Override
@@ -131,7 +143,7 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
public void register(final StateStore store,
final StateRestoreCallback stateRestoreCallback) {
- if (stores.containsKey(store.name())) {
+ if (globalStores.containsKey(store.name())) {
throw new IllegalArgumentException(String.format("Global Store %s has already been registered", store.name()));
}
@@ -173,7 +185,7 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
}
try {
restoreState(stateRestoreCallback, topicPartitions, highWatermarks, store.name());
- stores.put(store.name(), store);
+ globalStores.put(store.name(), store);
} finally {
globalConsumer.unsubscribe();
}
@@ -249,17 +261,27 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
long restoreCount = 0L;
while (offset < highWatermark) {
- final ConsumerRecords<byte[], byte[]> records = globalConsumer.poll(100);
- final List<KeyValue<byte[], byte[]>> restoreRecords = new ArrayList<>();
- for (ConsumerRecord<byte[], byte[]> record : records) {
- if (record.key() != null) {
- restoreRecords.add(KeyValue.pair(record.key(), record.value()));
+ try {
+ final ConsumerRecords<byte[], byte[]> records = globalConsumer.poll(100);
+ final List<KeyValue<byte[], byte[]>> restoreRecords = new ArrayList<>();
+ for (ConsumerRecord<byte[], byte[]> record : records) {
+ 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();
+ } catch (final InvalidOffsetException recoverableException) {
+ log.warn("Restoring GlobalStore {} failed due to: {}. Deleting global store to recreate from scratch.",
+ storeName,
+ recoverableException.getMessage());
+ reinitializeStateStoresForPartitions(recoverableException.partitions(), processorContext);
+
+ stateRestoreListener.onRestoreStart(topicPartition, storeName, offset, highWatermark);
+ restoreCount = 0L;
}
- stateRestoreAdapter.restoreAll(restoreRecords);
- stateRestoreListener.onBatchRestored(topicPartition, storeName, offset, restoreRecords.size());
- restoreCount += restoreRecords.size();
}
stateRestoreListener.onRestoreEnd(topicPartition, storeName, restoreCount);
checkpointableOffsets.put(topicPartition, offset);
@@ -268,8 +290,8 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
@Override
public void flush() {
- log.debug("Flushing all global stores registered in the state manager");
- for (StateStore store : this.stores.values()) {
+ log.debug("Flushing all global globalStores registered in the state manager");
+ for (StateStore store : this.globalStores.values()) {
try {
log.trace("Flushing global store={}", store.name());
store.flush();
@@ -283,11 +305,11 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
@Override
public void close(final Map<TopicPartition, Long> offsets) throws IOException {
try {
- if (stores.isEmpty()) {
+ if (globalStores.isEmpty()) {
return;
}
final StringBuilder closeFailed = new StringBuilder();
- for (final Map.Entry<String, StateStore> entry : stores.entrySet()) {
+ for (final Map.Entry<String, StateStore> entry : globalStores.entrySet()) {
log.debug("Closing global storage engine {}", entry.getKey());
try {
entry.getValue().close();
@@ -300,9 +322,9 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
.append("\n");
}
}
- stores.clear();
+ globalStores.clear();
if (closeFailed.length() > 0) {
- throw new ProcessorStateException("Exceptions caught during close of 1 or more global state stores\n" + closeFailed);
+ throw new ProcessorStateException("Exceptions caught during close of 1 or more global state globalStores\n" + closeFailed);
}
checkpoint(offsets);
} finally {
@@ -317,7 +339,7 @@ public class GlobalStateManagerImpl implements GlobalStateManager {
try {
checkpoint.write(checkpointableOffsets);
} catch (IOException e) {
- log.warn("Failed to write offsets checkpoint for global stores", e);
+ log.warn("Failed to write offsets checkpoint for global globalStores", e);
}
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
index 849af57..c18f3c7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
@@ -56,8 +56,9 @@ public class GlobalStateUpdateTask implements GlobalStateMaintainer {
* @throws IllegalStateException If store gets registered after initialized is already finished
* @throws StreamsException if the store's change log does not contain the partition
*/
+ @Override
public Map<TopicPartition, Long> initialize() {
- final Set<String> storeNames = stateMgr.initialize(processorContext);
+ final Set<String> storeNames = stateMgr.initialize();
final Map<String, String> storeNameToTopic = topology.storeToChangelogTopic();
for (final String storeName : storeNames) {
final String sourceTopic = storeNameToTopic.get(storeName);
@@ -69,7 +70,6 @@ public class GlobalStateUpdateTask implements GlobalStateMaintainer {
return stateMgr.checkpointed();
}
-
@SuppressWarnings("unchecked")
@Override
public void update(final ConsumerRecord<byte[], byte[]> record) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
index 9d202d1..1cc5c85 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
@@ -19,6 +19,7 @@ package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.utils.LogContext;
@@ -234,14 +235,20 @@ public class GlobalStreamThread extends Thread {
}
void pollAndUpdate() {
- final ConsumerRecords<byte[], byte[]> received = globalConsumer.poll(pollMs);
- for (ConsumerRecord<byte[], byte[]> record : received) {
- stateMaintainer.update(record);
- }
- final long now = time.milliseconds();
- if (flushInterval >= 0 && now >= lastFlush + flushInterval) {
- stateMaintainer.flushState();
- lastFlush = now;
+ try {
+ final ConsumerRecords<byte[], byte[]> received = globalConsumer.poll(pollMs);
+ for (ConsumerRecord<byte[], byte[]> record : received) {
+ stateMaintainer.update(record);
+ }
+ final long now = time.milliseconds();
+ if (flushInterval >= 0 && now >= lastFlush + flushInterval) {
+ stateMaintainer.flushState();
+ lastFlush = now;
+ }
+ } catch (final InvalidOffsetException recoverableException) {
+ log.error("Updating global state failed. You can restart KafkaStreams to recover from this error.", recoverableException);
+ throw new StreamsException("Updating global state failed. " +
+ "You can restart KafkaStreams to recover from this error.", recoverableException);
}
}
@@ -308,15 +315,19 @@ public class GlobalStreamThread extends Thread {
stateDirectory,
stateRestoreListener,
config);
+
+ final GlobalProcessorContextImpl globalProcessorContext = new GlobalProcessorContextImpl(
+ config,
+ stateMgr,
+ streamsMetrics,
+ cache);
+ stateMgr.setGlobalProcessorContext(globalProcessorContext);
+
final StateConsumer stateConsumer
= new StateConsumer(this.logContext,
globalConsumer,
new GlobalStateUpdateTask(topology,
- new GlobalProcessorContextImpl(
- config,
- stateMgr,
- streamsMetrics,
- cache),
+ globalProcessorContext,
stateMgr,
config.defaultDeserializationExceptionHandler(),
logContext),
@@ -324,6 +335,7 @@ public class GlobalStreamThread extends Thread {
config.getLong(StreamsConfig.POLL_MS_CONFIG),
config.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG));
stateConsumer.initialize();
+
return stateConsumer;
} catch (final LockException fatalException) {
final String errorMsg = "Could not lock global state directory. This could happen if multiple KafkaStreams " +
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalProcessorContext.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalProcessorContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalProcessorContext.java
index 57bb3ac..25df826 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalProcessorContext.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalProcessorContext.java
@@ -50,7 +50,12 @@ public interface InternalProcessorContext extends ProcessorContext {
ThreadCache getCache();
/**
- * Mark this contex as being initialized
+ * Mark this context as being initialized
*/
void initialized();
+
+ /**
+ * Mark this context as being uninitialized
+ */
+ void uninitialize();
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
----------------------------------------------------------------------
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 3a2803e..1ee0e14 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
@@ -33,27 +33,20 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-public class ProcessorStateManager implements StateManager {
-
+public class ProcessorStateManager extends AbstractStateManager {
private static final String STATE_CHANGELOG_TOPIC_SUFFIX = "-changelog";
- static final String CHECKPOINT_FILE_NAME = ".checkpoint";
private final Logger log;
- private final File baseDir;
private final TaskId taskId;
private final String logPrefix;
private final boolean isStandby;
private final ChangelogReader changelogReader;
- private final Map<String, StateStore> stores;
- private final Map<String, StateStore> globalStores;
private final Map<TopicPartition, Long> offsetLimits;
private final Map<TopicPartition, Long> restoredOffsets;
- private final Map<TopicPartition, Long> checkpointedOffsets;
private final Map<String, StateRestoreCallback> restoreCallbacks; // used for standby tasks, keyed by state topic name
private final Map<String, String> storeToChangelogTopic;
private final List<TopicPartition> changelogPartitions = new ArrayList<>();
@@ -61,7 +54,6 @@ public class ProcessorStateManager implements StateManager {
// TODO: this map does not work with customized grouper where multiple partitions
// of the same topic can be assigned to the same topic.
private final Map<String, TopicPartition> partitionForTopic;
- private OffsetCheckpoint checkpoint;
/**
* @throws ProcessorStateException if the task directory does not exist and could not be created
@@ -75,28 +67,25 @@ public class ProcessorStateManager implements StateManager {
final ChangelogReader changelogReader,
final boolean eosEnabled,
final LogContext logContext) throws IOException {
+ super(stateDirectory.directoryForTask(taskId));
+
+ this.log = logContext.logger(ProcessorStateManager.class);
this.taskId = taskId;
this.changelogReader = changelogReader;
logPrefix = String.format("task [%s] ", taskId);
- this.log = logContext.logger(getClass());
partitionForTopic = new HashMap<>();
for (final TopicPartition source : sources) {
partitionForTopic.put(source.topic(), source);
}
- stores = new LinkedHashMap<>();
- globalStores = new HashMap<>();
offsetLimits = new HashMap<>();
restoredOffsets = new HashMap<>();
this.isStandby = isStandby;
restoreCallbacks = isStandby ? new HashMap<String, StateRestoreCallback>() : null;
this.storeToChangelogTopic = storeToChangelogTopic;
- baseDir = stateDirectory.directoryForTask(taskId);
-
// load the checkpoint information
- checkpoint = new OffsetCheckpoint(new File(baseDir, CHECKPOINT_FILE_NAME));
- checkpointedOffsets = new HashMap<>(checkpoint.read());
+ checkpointableOffsets.putAll(checkpoint.read());
if (eosEnabled) {
// delete the checkpoint file after finish loading its stored offsets
@@ -120,42 +109,55 @@ public class ProcessorStateManager implements StateManager {
@Override
public void register(final StateStore store,
final StateRestoreCallback stateRestoreCallback) {
- log.debug("Registering state store {} to its state manager", store.name());
+ final String storeName = store.name();
+ log.debug("Registering state store {} to its state manager", storeName);
- if (store.name().equals(CHECKPOINT_FILE_NAME)) {
+ if (CHECKPOINT_FILE_NAME.equals(storeName)) {
throw new IllegalArgumentException(String.format("%sIllegal store name: %s", logPrefix, CHECKPOINT_FILE_NAME));
}
- if (stores.containsKey(store.name())) {
- throw new IllegalArgumentException(String.format("%sStore %s has already been registered.", logPrefix, store.name()));
+ if (stores.containsKey(storeName)) {
+ throw new IllegalArgumentException(String.format("%sStore %s has already been registered.", logPrefix, storeName));
}
// check that the underlying change log topic exist or not
- final String topic = storeToChangelogTopic.get(store.name());
+ final String topic = storeToChangelogTopic.get(storeName);
if (topic == null) {
- stores.put(store.name(), store);
+ stores.put(storeName, store);
return;
}
final TopicPartition storePartition = new TopicPartition(topic, getPartition(topic));
if (isStandby) {
- log.trace("Preparing standby replica of state store {} with changelog topic {}", store.name(), topic);
+ log.trace("Preparing standby replica of persistent state store {} with changelog topic {}", storeName, topic);
restoreCallbacks.put(topic, stateRestoreCallback);
+
} else {
- log.trace("Restoring state store {} from changelog topic {}", store.name(), topic);
+ log.trace("Restoring state store {} from changelog topic {}", storeName, topic);
final StateRestorer restorer = new StateRestorer(storePartition,
new CompositeRestoreListener(stateRestoreCallback),
- checkpointedOffsets.get(storePartition),
+ checkpointableOffsets.get(storePartition),
offsetLimit(storePartition),
store.persistent(),
- store.name());
+ storeName);
changelogReader.register(restorer);
}
changelogPartitions.add(storePartition);
- stores.put(store.name(), store);
+ stores.put(storeName, store);
+ }
+
+ @Override
+ public void reinitializeStateStoresForPartitions(final Collection<TopicPartition> partitions,
+ final InternalProcessorContext processorContext) {
+ super.reinitializeStateStoresForPartitions(
+ log,
+ stores,
+ storeToChangelogTopic,
+ partitions,
+ processorContext);
}
@Override
@@ -167,8 +169,8 @@ public class ProcessorStateManager implements StateManager {
final int partition = getPartition(topicName);
final TopicPartition storePartition = new TopicPartition(topicName, partition);
- if (checkpointedOffsets.containsKey(storePartition)) {
- partitionsAndOffsets.put(storePartition, checkpointedOffsets.get(storePartition));
+ if (checkpointableOffsets.containsKey(storePartition)) {
+ partitionsAndOffsets.put(storePartition, checkpointableOffsets.get(storePartition));
} else {
partitionsAndOffsets.put(storePartition, -1L);
}
@@ -281,7 +283,7 @@ public class ProcessorStateManager implements StateManager {
if (ackedOffsets != null) {
checkpoint(ackedOffsets);
}
-
+ stores.clear();
}
if (firstException != null) {
@@ -293,7 +295,7 @@ public class ProcessorStateManager implements StateManager {
@Override
public void checkpoint(final Map<TopicPartition, Long> ackedOffsets) {
log.trace("Writing checkpoint: {}", ackedOffsets);
- checkpointedOffsets.putAll(changelogReader.restoredOffsets());
+ checkpointableOffsets.putAll(changelogReader.restoredOffsets());
for (final StateStore store : stores.values()) {
final String storeName = store.name();
// only checkpoint the offset to the offsets file if
@@ -303,9 +305,9 @@ public class ProcessorStateManager implements StateManager {
final TopicPartition topicPartition = new TopicPartition(changelogTopic, getPartition(storeName));
if (ackedOffsets.containsKey(topicPartition)) {
// store the last offset + 1 (the log position after restoration)
- checkpointedOffsets.put(topicPartition, ackedOffsets.get(topicPartition) + 1);
+ checkpointableOffsets.put(topicPartition, ackedOffsets.get(topicPartition) + 1);
} else if (restoredOffsets.containsKey(topicPartition)) {
- checkpointedOffsets.put(topicPartition, restoredOffsets.get(topicPartition));
+ checkpointableOffsets.put(topicPartition, restoredOffsets.get(topicPartition));
}
}
}
@@ -314,7 +316,7 @@ public class ProcessorStateManager implements StateManager {
if (checkpoint == null) {
checkpoint = new OffsetCheckpoint(new File(baseDir, CHECKPOINT_FILE_NAME));
}
- checkpoint.write(checkpointedOffsets);
+ checkpoint.write(checkpointableOffsets);
} catch (final IOException e) {
log.warn("Failed to write checkpoint file to {}:", new File(baseDir, CHECKPOINT_FILE_NAME), e);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
index 73fbf63..837f607 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
@@ -131,8 +131,10 @@ public class StandbyTask extends AbstractTask {
log.debug("Closing");
boolean committedSuccessfully = false;
try {
- commit();
- committedSuccessfully = true;
+ if (clean) {
+ commit();
+ committedSuccessfully = true;
+ }
} finally {
closeStateManager(committedSuccessfully);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManager.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManager.java
index 2a8d9a3..f6efde6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManager.java
@@ -23,6 +23,7 @@ import org.apache.kafka.streams.processor.StateStore;
import java.io.File;
import java.io.IOException;
+import java.util.Collection;
import java.util.Map;
interface StateManager extends Checkpointable {
@@ -37,7 +38,10 @@ interface StateManager extends Checkpointable {
void flush();
- void close(Map<TopicPartition, Long> offsets) throws IOException;
+ void reinitializeStateStoresForPartitions(final Collection<TopicPartition> partitions,
+ final InternalProcessorContext processorContext);
+
+ void close(final Map<TopicPartition, Long> offsets) throws IOException;
StateStore getGlobalStore(final String name);
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
----------------------------------------------------------------------
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 83c783d..178d2bb 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
@@ -19,6 +19,7 @@ package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.TimeoutException;
@@ -78,12 +79,24 @@ public class StoreChangelogReader implements ChangelogReader {
return completed();
}
- final Set<TopicPartition> partitions = new HashSet<>(needsRestoring.keySet());
- final ConsumerRecords<byte[], byte[]> allRecords = restoreConsumer.poll(10);
- for (final TopicPartition partition : partitions) {
- restorePartition(allRecords, partition, active.restoringTaskFor(partition));
+ final Set<TopicPartition> restoringPartitions = new HashSet<>(needsRestoring.keySet());
+ try {
+ final ConsumerRecords<byte[], byte[]> allRecords = restoreConsumer.poll(10);
+ for (final TopicPartition partition : restoringPartitions) {
+ restorePartition(allRecords, partition, active.restoringTaskFor(partition));
+ }
+ } catch (final InvalidOffsetException recoverableException) {
+ log.warn("Restoring StreamTasks failed. Deleting StreamTasks stores to recreate from scratch.", recoverableException);
+ final Set<TopicPartition> partitions = recoverableException.partitions();
+ for (final TopicPartition partition : partitions) {
+ final StreamTask task = active.restoringTaskFor(partition);
+ log.info("Reinitializing StreamTask {}", task);
+ task.reinitializeStateStoresForPartitions(recoverableException.partitions());
+ }
+ restoreConsumer.seekToBeginning(partitions);
}
+
if (needsRestoring.isEmpty()) {
restoreConsumer.unsubscribe();
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index 55456d0..f2fa448 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -33,7 +33,6 @@ import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskMigratedException;
import org.apache.kafka.streams.processor.Cancellable;
-import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.processor.TaskId;
@@ -641,11 +640,6 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator
}
// visible for testing only
- ProcessorContext processorContext() {
- return processorContext;
- }
-
- // visible for testing only
RecordCollector recordCollector() {
return recordCollector;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index a9786f9..696081d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -1036,22 +1036,33 @@ public class StreamThread extends Thread {
processStandbyRecords = false;
}
- final ConsumerRecords<byte[], byte[]> records = restoreConsumer.poll(0);
+ try {
+ final ConsumerRecords<byte[], byte[]> records = restoreConsumer.poll(0);
- if (!records.isEmpty()) {
- for (final TopicPartition partition : records.partitions()) {
- final StandbyTask task = taskManager.standbyTask(partition);
+ if (!records.isEmpty()) {
+ for (final TopicPartition partition : records.partitions()) {
+ final StandbyTask task = taskManager.standbyTask(partition);
- if (task == null) {
- throw new StreamsException(logPrefix + "Missing standby task for partition " + partition);
- }
+ if (task == null) {
+ throw new StreamsException(logPrefix + "Missing standby task for partition " + partition);
+ }
- final List<ConsumerRecord<byte[], byte[]>> remaining = task.update(partition, records.records(partition));
- if (remaining != null) {
- restoreConsumer.pause(singleton(partition));
- standbyRecords.put(partition, remaining);
+ final List<ConsumerRecord<byte[], byte[]>> remaining = task.update(partition, records.records(partition));
+ if (remaining != null) {
+ restoreConsumer.pause(singleton(partition));
+ standbyRecords.put(partition, remaining);
+ }
}
}
+ } catch (final InvalidOffsetException recoverableException) {
+ log.warn("Updating StandbyTasks failed. Deleting StandbyTasks stores to recreate from scratch.", recoverableException);
+ final Set<TopicPartition> partitions = recoverableException.partitions();
+ for (final TopicPartition partition : partitions) {
+ final StandbyTask task = taskManager.standbyTask(partition);
+ log.info("Reinitializing StandbyTask {}", task);
+ task.reinitializeStateStoresForPartitions(recoverableException.partitions());
+ }
+ restoreConsumer.seekToBeginning(partitions);
}
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/state/internals/InnerMeteredKeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InnerMeteredKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InnerMeteredKeyValueStore.java
index 5ff8a26..a34851a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/InnerMeteredKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/InnerMeteredKeyValueStore.java
@@ -159,7 +159,6 @@ class InnerMeteredKeyValueStore<K, IK, V, IV> extends WrappedStateStore.Abstract
} else {
inner.init(InnerMeteredKeyValueStore.this.context, InnerMeteredKeyValueStore.this.root);
}
-
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStore.java
index a6ff8d5..35647b7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStore.java
@@ -102,8 +102,6 @@ public class MeteredKeyValueBytesStore<K, V> extends WrappedStateStore.AbstractS
keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
valueSerde == null ? (Serde<V>) context.valueSerde() : valueSerde);
innerMetered.init(context, root);
-
-
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
index a06f7e8..12db711 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
@@ -190,11 +190,9 @@ public class StreamsConfigTest {
@Test
public void shouldSupportPrefixedRestoreConsumerConfigs() {
- props.put(consumerPrefix(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), "earliest");
props.put(consumerPrefix(ConsumerConfig.METRICS_NUM_SAMPLES_CONFIG), 1);
final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> consumerConfigs = streamsConfig.getRestoreConsumerConfigs("clientId");
- assertEquals("earliest", consumerConfigs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));
assertEquals(1, consumerConfigs.get(ConsumerConfig.METRICS_NUM_SAMPLES_CONFIG));
}
@@ -245,11 +243,9 @@ public class StreamsConfigTest {
@Test
public void shouldBeSupportNonPrefixedRestoreConsumerConfigs() {
- props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.METRICS_NUM_SAMPLES_CONFIG, 1);
final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> consumerConfigs = streamsConfig.getRestoreConsumerConfigs("groupId");
- assertEquals("earliest", consumerConfigs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));
assertEquals(1, consumerConfigs.get(ConsumerConfig.METRICS_NUM_SAMPLES_CONFIG));
}
@@ -305,11 +301,9 @@ public class StreamsConfigTest {
@Test
public void shouldOverrideStreamsDefaultConsumerConifgsOnRestoreConsumer() {
- props.put(StreamsConfig.consumerPrefix(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), "latest");
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "10");
final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> consumerConfigs = streamsConfig.getRestoreConsumerConfigs("clientId");
- assertEquals("latest", consumerConfigs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));
assertEquals("10", consumerConfigs.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/04395175/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
index e75b54f..776110c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
@@ -25,52 +25,69 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.LockException;
import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.test.MockProcessorContext;
+import org.apache.kafka.test.MockRestoreCallback;
import org.apache.kafka.test.MockStateRestoreListener;
import org.apache.kafka.test.TestUtils;
import org.easymock.EasyMock;
import org.junit.Before;
import org.junit.Test;
+import java.io.File;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
-import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Properties;
+import static org.easymock.EasyMock.expect;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class AbstractTaskTest {
private final TaskId id = new TaskId(0, 0);
private StateDirectory stateDirectory = EasyMock.createMock(StateDirectory.class);
+ private final TopicPartition storeTopicPartition1 = new TopicPartition("t1", 0);
+ private final TopicPartition storeTopicPartition2 = new TopicPartition("t2", 0);
+ private final TopicPartition storeTopicPartition3 = new TopicPartition("t3", 0);
+ private final TopicPartition storeTopicPartition4 = new TopicPartition("t4", 0);
+ private final Collection<TopicPartition> storeTopicPartitions
+ = Utils.mkSet(storeTopicPartition1, storeTopicPartition2, storeTopicPartition3, storeTopicPartition4);
@Before
public void before() {
- EasyMock.expect(stateDirectory.directoryForTask(id)).andReturn(TestUtils.tempDirectory());
+ expect(stateDirectory.directoryForTask(id)).andReturn(TestUtils.tempDirectory());
}
@Test(expected = ProcessorStateException.class)
public void shouldThrowProcessorStateExceptionOnInitializeOffsetsWhenAuthorizationException() {
final Consumer consumer = mockConsumer(new AuthorizationException("blah"));
- final AbstractTask task = createTask(consumer, Collections.<StateStore>emptyList());
+ final AbstractTask task = createTask(consumer, Collections.<StateStore, String>emptyMap());
task.updateOffsetLimits();
}
@Test(expected = ProcessorStateException.class)
public void shouldThrowProcessorStateExceptionOnInitializeOffsetsWhenKafkaException() {
final Consumer consumer = mockConsumer(new KafkaException("blah"));
- final AbstractTask task = createTask(consumer, Collections.<StateStore>emptyList());
+ final AbstractTask task = createTask(consumer, Collections.<StateStore, String>emptyMap());
task.updateOffsetLimits();
}
@Test(expected = WakeupException.class)
public void shouldThrowWakeupExceptionOnInitializeOffsetsWhenWakeupException() {
final Consumer consumer = mockConsumer(new WakeupException());
- final AbstractTask task = createTask(consumer, Collections.<StateStore>emptyList());
+ final AbstractTask task = createTask(consumer, Collections.<StateStore, String>emptyMap());
task.updateOffsetLimits();
}
@@ -78,10 +95,10 @@ public class AbstractTaskTest {
public void shouldThrowLockExceptionIfFailedToLockStateDirectoryWhenTopologyHasStores() throws IOException {
final Consumer consumer = EasyMock.createNiceMock(Consumer.class);
final StateStore store = EasyMock.createNiceMock(StateStore.class);
- EasyMock.expect(stateDirectory.lock(id)).andReturn(false);
+ expect(stateDirectory.lock(id)).andReturn(false);
EasyMock.replay(stateDirectory);
- final AbstractTask task = createTask(consumer, Collections.singletonList(store));
+ final AbstractTask task = createTask(consumer, Collections.singletonMap(store, "dummy"));
try {
task.initializeStateStores();
@@ -93,11 +110,11 @@ public class AbstractTaskTest {
}
@Test
- public void shouldNotAttemptToLockIfNoStores() throws IOException {
+ public void shouldNotAttemptToLockIfNoStores() {
final Consumer consumer = EasyMock.createNiceMock(Consumer.class);
EasyMock.replay(stateDirectory);
- final AbstractTask task = createTask(consumer, Collections.<StateStore>emptyList());
+ final AbstractTask task = createTask(consumer, Collections.<StateStore, String>emptyMap());
task.initializeStateStores();
@@ -105,20 +122,122 @@ public class AbstractTaskTest {
EasyMock.verify(stateDirectory);
}
+ @Test
+ public void shouldDeleteAndRecreateStoreDirectoryOnReinitialize() throws IOException {
+ final StreamsConfig streamsConfig = new StreamsConfig(new Properties() {
+ {
+ put(StreamsConfig.APPLICATION_ID_CONFIG, "app-id");
+ put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+ put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
+ }
+ });
+ final Consumer consumer = EasyMock.createNiceMock(Consumer.class);
+
+ final StateStore store1 = EasyMock.createNiceMock(StateStore.class);
+ final StateStore store2 = EasyMock.createNiceMock(StateStore.class);
+ final StateStore store3 = EasyMock.createNiceMock(StateStore.class);
+ final StateStore store4 = EasyMock.createNiceMock(StateStore.class);
+ final String storeName1 = "storeName1";
+ final String storeName2 = "storeName2";
+ final String storeName3 = "storeName3";
+ final String storeName4 = "storeName4";
+
+ expect(store1.name()).andReturn(storeName1).anyTimes();
+ EasyMock.replay(store1);
+ expect(store2.name()).andReturn(storeName2).anyTimes();
+ EasyMock.replay(store2);
+ expect(store3.name()).andReturn(storeName3).anyTimes();
+ EasyMock.replay(store3);
+ expect(store4.name()).andReturn(storeName4).anyTimes();
+ EasyMock.replay(store4);
+
+ final StateDirectory stateDirectory = new StateDirectory(streamsConfig, new MockTime());
+ final AbstractTask task = createTask(
+ consumer,
+ new HashMap<StateStore, String>() {
+ {
+ put(store1, storeTopicPartition1.topic());
+ put(store2, storeTopicPartition2.topic());
+ put(store3, storeTopicPartition3.topic());
+ put(store4, storeTopicPartition4.topic());
+ }
+ },
+ stateDirectory);
+
+ final String taskDir = stateDirectory.directoryForTask(task.id).getAbsolutePath();
+ final File storeDirectory1 = new File(taskDir
+ + File.separator + "rocksdb"
+ + File.separator + storeName1);
+ final File storeDirectory2 = new File(taskDir
+ + File.separator + "rocksdb"
+ + File.separator + storeName2);
+ final File storeDirectory3 = new File(taskDir
+ + File.separator + storeName3);
+ final File storeDirectory4 = new File(taskDir
+ + File.separator + storeName4);
+ final File testFile1 = new File(storeDirectory1.getAbsolutePath() + File.separator + "testFile");
+ final File testFile2 = new File(storeDirectory2.getAbsolutePath() + File.separator + "testFile");
+ final File testFile3 = new File(storeDirectory3.getAbsolutePath() + File.separator + "testFile");
+ final File testFile4 = new File(storeDirectory4.getAbsolutePath() + File.separator + "testFile");
+
+ storeDirectory1.mkdirs();
+ storeDirectory2.mkdirs();
+ storeDirectory3.mkdirs();
+ storeDirectory4.mkdirs();
+
+ testFile1.createNewFile();
+ assertTrue(testFile1.exists());
+ testFile2.createNewFile();
+ assertTrue(testFile2.exists());
+ testFile3.createNewFile();
+ assertTrue(testFile3.exists());
+ testFile4.createNewFile();
+ assertTrue(testFile4.exists());
+
+ task.processorContext = new MockProcessorContext(stateDirectory.directoryForTask(task.id), streamsConfig);
+
+ task.stateMgr.register(store1, new MockRestoreCallback());
+ task.stateMgr.register(store2, new MockRestoreCallback());
+ task.stateMgr.register(store3, new MockRestoreCallback());
+ task.stateMgr.register(store4, new MockRestoreCallback());
+
+ // only reinitialize store1 and store3 -- store2 and store4 should be untouched
+ task.reinitializeStateStoresForPartitions(Utils.mkSet(storeTopicPartition1, storeTopicPartition3));
+
+ assertFalse(testFile1.exists());
+ assertTrue(testFile2.exists());
+ assertFalse(testFile3.exists());
+ assertTrue(testFile4.exists());
+ }
+
+ private AbstractTask createTask(final Consumer consumer,
+ final Map<StateStore, String> stateStoresToChangelogTopics) {
+ return createTask(consumer, stateStoresToChangelogTopics, stateDirectory);
+ }
+
@SuppressWarnings("unchecked")
- private AbstractTask createTask(final Consumer consumer, final List<StateStore> stateStores) {
+ private AbstractTask createTask(final Consumer consumer,
+ final Map<StateStore, String> stateStoresToChangelogTopics,
+ final StateDirectory stateDirectory) {
final Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "app");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummyhost:9092");
final StreamsConfig config = new StreamsConfig(properties);
+
+ final Map<String, String> storeNamesToChangelogTopics = new HashMap<>(stateStoresToChangelogTopics.size());
+ for (final Map.Entry<StateStore, String> e : stateStoresToChangelogTopics.entrySet()) {
+ storeNamesToChangelogTopics.put(e.getKey().name(), e.getValue());
+ }
+
return new AbstractTask(id,
- Collections.singletonList(new TopicPartition("t", 0)),
- ProcessorTopology.withLocalStores(stateStores, Collections.<String, String>emptyMap()),
- (Consumer<byte[], byte[]>) consumer,
- new StoreChangelogReader((Consumer<byte[], byte[]>) consumer, new MockStateRestoreListener(), new LogContext("stream-task-test ")),
+ storeTopicPartitions,
+ ProcessorTopology.withLocalStores(new ArrayList<>(stateStoresToChangelogTopics.keySet()), storeNamesToChangelogTopics),
+ consumer,
+ new StoreChangelogReader(consumer, new MockStateRestoreListener(), new LogContext("stream-task-test ")),
false,
stateDirectory,
config) {
+
@Override
public void resume() {}