From 386c42869d23108856cc426effc7e63438043d86 Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Tue, 16 Jun 2020 19:39:32 -0700 Subject: [PATCH] KAFKA-9891: add integration tests for EOS and StandbyTask --- .../kafka/streams/StoreQueryParameters.java | 2 +- .../StandbyTaskEOSIntegrationTest.java | 230 +++++++++++++++++- 2 files changed, 224 insertions(+), 8 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/StoreQueryParameters.java b/streams/src/main/java/org/apache/kafka/streams/StoreQueryParameters.java index 332ac1a634956..3789e6fc8c1ef 100644 --- a/streams/src/main/java/org/apache/kafka/streams/StoreQueryParameters.java +++ b/streams/src/main/java/org/apache/kafka/streams/StoreQueryParameters.java @@ -38,7 +38,7 @@ private StoreQueryParameters(final String storeName, final QueryableStoreType } public static StoreQueryParameters fromNameAndType(final String storeName, - final QueryableStoreType queryableStoreType) { + final QueryableStoreType queryableStoreType) { return new StoreQueryParameters(storeName, queryableStoreType, null, false); } diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java index 4c7602e4f541c..0cf351ea929e2 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java @@ -18,18 +18,25 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.IntegerSerializer; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.StoreQueryParameters; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster; import org.apache.kafka.streams.integration.utils.IntegrationTestUtils; import org.apache.kafka.streams.kstream.Consumed; +import org.apache.kafka.streams.kstream.Transformer; +import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.processor.internals.StateDirectory; +import org.apache.kafka.streams.state.KeyValueStore; +import org.apache.kafka.streams.state.QueryableStoreTypes; +import org.apache.kafka.streams.state.Stores; import org.apache.kafka.streams.state.internals.OffsetCheckpoint; import org.apache.kafka.test.TestUtils; import org.junit.Before; @@ -41,17 +48,18 @@ import org.junit.runners.Parameterized; import java.io.File; -import java.io.IOException; import java.time.Duration; import java.util.Collection; import java.util.Collections; import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import static java.util.Arrays.asList; import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName; import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning; +import static org.apache.kafka.test.TestUtils.waitForCondition; import static org.junit.Assert.assertTrue; /** @@ -61,6 +69,10 @@ @RunWith(Parameterized.class) public class StandbyTaskEOSIntegrationTest { + private final static long REBALANCE_TIMEOUT = Duration.ofMinutes(2L).toMillis(); + private final static int KEY_0 = 0; + private final static int KEY_1 = 1; + @Parameterized.Parameters(name = "{0}") public static Collection data() { return asList(new String[][] { @@ -72,8 +84,12 @@ public static Collection data() { @Parameterized.Parameter public String eosConfig; + private final AtomicBoolean skipRecord = new AtomicBoolean(false); + private String appId; private String inputTopic; + private String storeName; + private String outputTopic; @ClassRule public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(3); @@ -86,8 +102,11 @@ public void createTopics() throws Exception { final String safeTestName = safeUniqueTestName(getClass(), testName); appId = "app-" + safeTestName; inputTopic = "input-" + safeTestName; - CLUSTER.deleteTopicsAndWait(inputTopic, appId + "-KSTREAM-AGGREGATE-STATE-STORE-0000000001-changelog"); + outputTopic = "output-" + safeTestName; + storeName = "store-" + safeTestName; + CLUSTER.deleteTopicsAndWait(inputTopic, outputTopic, appId + "-KSTREAM-AGGREGATE-STATE-STORE-0000000001-changelog"); CLUSTER.createTopic(inputTopic, 1, 3); + CLUSTER.createTopic(outputTopic, 1, 3); } @Test @@ -95,13 +114,16 @@ public void shouldSurviveWithOneTaskAsStandby() throws Exception { IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp( inputTopic, Collections.singletonList( - new KeyValue<>(0, 0)), + new KeyValue<>(0, 0) + ), TestUtils.producerConfig( CLUSTER.bootstrapServers(), IntegerSerializer.class, IntegerSerializer.class, - new Properties()), - 10L); + new Properties() + ), + 10L + ); final String stateDirPath = TestUtils.tempDirectory(appId).getPath(); @@ -109,7 +131,7 @@ public void shouldSurviveWithOneTaskAsStandby() throws Exception { try ( final KafkaStreams streamInstanceOne = buildStreamWithDirtyStateDir(stateDirPath + "/" + appId + "-1/", instanceLatch); - final KafkaStreams streamInstanceTwo = buildStreamWithDirtyStateDir(stateDirPath + "/" + appId + "-2/", instanceLatch); + final KafkaStreams streamInstanceTwo = buildStreamWithDirtyStateDir(stateDirPath + "/" + appId + "-2/", instanceLatch) ) { startApplicationAndWaitUntilRunning(asList(streamInstanceOne, streamInstanceTwo), Duration.ofSeconds(60)); @@ -125,7 +147,7 @@ public void shouldSurviveWithOneTaskAsStandby() throws Exception { } private KafkaStreams buildStreamWithDirtyStateDir(final String stateDirPath, - final CountDownLatch recordProcessLatch) throws IOException { + final CountDownLatch recordProcessLatch) throws Exception { final StreamsBuilder builder = new StreamsBuilder(); final TaskId taskId = new TaskId(0, 0); @@ -151,6 +173,198 @@ private KafkaStreams buildStreamWithDirtyStateDir(final String stateDirPath, return new KafkaStreams(builder.build(), props); } + @Test + public void shouldWipeOutStandbyStateDirectoryIfCheckpointIsMissing() throws Exception { + final String base = TestUtils.tempDirectory(appId).getPath(); + + IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp( + inputTopic, + Collections.singletonList( + new KeyValue<>(KEY_0, 0) + ), + TestUtils.producerConfig( + CLUSTER.bootstrapServers(), + IntegerSerializer.class, + IntegerSerializer.class, + new Properties() + ), + 10L + ); + + try ( + final KafkaStreams streamInstanceOne = buildWithDeduplicationTopology(base + "-1"); + final KafkaStreams streamInstanceTwo = buildWithDeduplicationTopology(base + "-2"); + final KafkaStreams streamInstanceOneRecovery = buildWithDeduplicationTopology(base + "-1") + ) { + // start first instance and wait for processing + startApplicationAndWaitUntilRunning(Collections.singletonList(streamInstanceOne), Duration.ofSeconds(30)); + IntegrationTestUtils.waitUntilMinRecordsReceived( + TestUtils.consumerConfig( + CLUSTER.bootstrapServers(), + IntegerDeserializer.class, + IntegerDeserializer.class + ), + outputTopic, + 1 + ); + + // start second instance and wait for standby replication + startApplicationAndWaitUntilRunning(Collections.singletonList(streamInstanceTwo), Duration.ofSeconds(30)); + waitForCondition( + () -> streamInstanceTwo.store( + StoreQueryParameters.fromNameAndType( + storeName, + QueryableStoreTypes.keyValueStore() + ).enableStaleStores() + ).get(KEY_0) != null, + REBALANCE_TIMEOUT, + "Could not get key from standby store" + ); + // sanity check that first instance is still active + waitForCondition( + () -> streamInstanceOne.store( + StoreQueryParameters.fromNameAndType( + storeName, + QueryableStoreTypes.keyValueStore() + ) + ).get(KEY_0) != null, + "Could not get key from main store" + ); + + // inject poison pill and wait for crash of first instance and recovery on second instance + IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp( + inputTopic, + Collections.singletonList( + new KeyValue<>(KEY_1, 0) + ), + TestUtils.producerConfig( + CLUSTER.bootstrapServers(), + IntegerSerializer.class, + IntegerSerializer.class, + new Properties() + ), + 10L + ); + waitForCondition( + () -> streamInstanceOne.state() == KafkaStreams.State.ERROR, + "Stream instance 1 did not go into error state" + ); + streamInstanceOne.close(); + + IntegrationTestUtils.waitUntilMinRecordsReceived( + TestUtils.consumerConfig( + CLUSTER.bootstrapServers(), + IntegerDeserializer.class, + IntegerDeserializer.class + ), + outputTopic, + 2 + ); + + // "restart" first client and wait for standby recovery + // (could actually also be active, but it does not matter as long as we enable "state stores" + startApplicationAndWaitUntilRunning( + Collections.singletonList(streamInstanceOneRecovery), + Duration.ofSeconds(30) + ); + waitForCondition( + () -> streamInstanceOneRecovery.store( + StoreQueryParameters.fromNameAndType( + storeName, + QueryableStoreTypes.keyValueStore() + ).enableStaleStores() + ).get(KEY_0) != null, + "Could not get key from recovered standby store" + ); + + streamInstanceTwo.close(); + waitForCondition( + () -> streamInstanceOneRecovery.store( + StoreQueryParameters.fromNameAndType( + storeName, + QueryableStoreTypes.keyValueStore() + ) + ).get(KEY_0) != null, + REBALANCE_TIMEOUT, + "Could not get key from recovered main store" + ); + + // re-inject poison pill and wait for crash of first instance + skipRecord.set(false); + IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp( + inputTopic, + Collections.singletonList( + new KeyValue<>(KEY_1, 0) + ), + TestUtils.producerConfig( + CLUSTER.bootstrapServers(), + IntegerSerializer.class, + IntegerSerializer.class, + new Properties() + ), + 10L + ); + waitForCondition( + () -> streamInstanceOneRecovery.state() == KafkaStreams.State.ERROR, + "Stream instance 1 did not go into error state" + ); + } + } + + private KafkaStreams buildWithDeduplicationTopology(final String stateDirPath) { + final StreamsBuilder builder = new StreamsBuilder(); + + builder.addStateStore(Stores.keyValueStoreBuilder( + Stores.persistentKeyValueStore(storeName), + Serdes.Integer(), + Serdes.Integer()) + ); + builder.stream(inputTopic) + .transform( + () -> new Transformer>() { + private KeyValueStore store; + + @SuppressWarnings("unchecked") + @Override + public void init(final ProcessorContext context) { + store = (KeyValueStore) context.getStateStore(storeName); + } + + @Override + public KeyValue transform(final Integer key, final Integer value) { + if (skipRecord.get()) { + // we only forward so we can verify the skipping by reading the output topic + // the goal is skipping is to not modify the state store + return KeyValue.pair(key, value); + } + + if (store.get(key) != null) { + return null; + } + + store.put(key, value); + store.flush(); + + if (key == KEY_1) { + // after error injection, we need to avoid a consecutive error after rebalancing + skipRecord.set(true); + throw new RuntimeException("Injected test error"); + } + + return KeyValue.pair(key, value); + } + + @Override + public void close() { } + }, + storeName + ) + .to(outputTopic); + + return new KafkaStreams(builder.build(), props(stateDirPath)); + } + + private Properties props(final String stateDirPath) { final Properties streamsConfiguration = new Properties(); streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, appId); @@ -162,6 +376,8 @@ private Properties props(final String stateDirPath) { streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); + // need to set to zero to get predictable active/standby task assignments + streamsConfiguration.put(StreamsConfig.ACCEPTABLE_RECOVERY_LAG_CONFIG, 0); streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); return streamsConfiguration;