diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index f3c463457c01e..4e137eb51c793 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -156,7 +156,7 @@
files="(TopologyBuilder|KafkaStreams|KStreamImpl|KTableImpl|StreamThread|StreamTask).java"/>
+ files="(KTableImpl|StreamsPartitionAssignor|EosBetaUpgradeIntegrationTest).java"/>
@@ -176,12 +176,16 @@
files="StreamThread.java"/>
+
+
+ files="(AssignorConfiguration|EosBetaUpgradeIntegrationTest|InternalTopologyBuilder|KafkaStreams|ProcessorStateManager|StreamsPartitionAssignor|StreamThread|TaskManager).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 46888481766c6..80209131b0dda 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -828,6 +828,8 @@ public class StreamsConfig extends AbstractConfig {
static {
final Map tempProducerDefaultOverrides = new HashMap<>();
tempProducerDefaultOverrides.put(ProducerConfig.LINGER_MS_CONFIG, "100");
+ // Reduce the transaction timeout for quicker pending offset expiration on broker side.
+ tempProducerDefaultOverrides.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10000);
PRODUCER_DEFAULT_OVERRIDES = Collections.unmodifiableMap(tempProducerDefaultOverrides);
}
@@ -1273,9 +1275,6 @@ public Map getProducerConfigs(final String clientId) {
// add client id with stream client id prefix
props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
- // Reduce the transaction timeout for quicker pending offset expiration on broker side.
- props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10000);
-
return props;
}
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 0b4e614e6d5f7..e4dff94483f8a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
@@ -399,9 +399,11 @@ public void shouldOverrideStreamsDefaultConsumerConfigs() {
@Test
public void shouldOverrideStreamsDefaultProducerConfigs() {
props.put(StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG), "10000");
+ props.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), "30000");
final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map producerConfigs = streamsConfig.getProducerConfigs(clientId);
assertEquals("10000", producerConfigs.get(ProducerConfig.LINGER_MS_CONFIG));
+ assertEquals("30000", producerConfigs.get(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG));
}
@Test
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/EosBetaUpgradeIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/EosBetaUpgradeIntegrationTest.java
new file mode 100644
index 0000000000000..eff49919fca66
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/EosBetaUpgradeIntegrationTest.java
@@ -0,0 +1,1105 @@
+/*
+ * 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.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Partitioner;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Utils;
+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.KStream;
+import org.apache.kafka.streams.kstream.Transformer;
+import org.apache.kafka.streams.kstream.TransformerSupplier;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.QueryableStoreTypes;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.test.IntegrationTest;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestUtils;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.apache.kafka.test.TestUtils.waitForCondition;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
+
+@RunWith(Parameterized.class)
+@Category({IntegrationTest.class})
+public class EosBetaUpgradeIntegrationTest {
+
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection data() {
+ return Arrays.asList(new Boolean[][] {
+ {false},
+ {true}
+ });
+ }
+
+ @Parameterized.Parameter
+ public boolean injectError;
+
+ private static final int NUM_BROKERS = 3;
+ private static final int MAX_POLL_INTERVAL_MS = 100 * 1000;
+ private static final int MAX_WAIT_TIME_MS = 60 * 1000;
+
+ private static final List> TWO_REBALANCES_STARTUP =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ KeyValue.pair(KafkaStreams.State.CREATED, KafkaStreams.State.REBALANCING),
+ KeyValue.pair(KafkaStreams.State.REBALANCING, KafkaStreams.State.RUNNING),
+ KeyValue.pair(KafkaStreams.State.RUNNING, KafkaStreams.State.REBALANCING),
+ KeyValue.pair(KafkaStreams.State.REBALANCING, KafkaStreams.State.RUNNING)
+ )
+ );
+ private static final List> TWO_REBALANCES_RUNNING =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ KeyValue.pair(KafkaStreams.State.RUNNING, KafkaStreams.State.REBALANCING),
+ KeyValue.pair(KafkaStreams.State.REBALANCING, KafkaStreams.State.RUNNING),
+ KeyValue.pair(KafkaStreams.State.RUNNING, KafkaStreams.State.REBALANCING),
+ KeyValue.pair(KafkaStreams.State.REBALANCING, KafkaStreams.State.RUNNING)
+ )
+ );
+ private static final List> SINGLE_REBALANCE =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ KeyValue.pair(KafkaStreams.State.RUNNING, KafkaStreams.State.REBALANCING),
+ KeyValue.pair(KafkaStreams.State.REBALANCING, KafkaStreams.State.RUNNING)
+ )
+ );
+ private static final List> CLOSE =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ KeyValue.pair(KafkaStreams.State.RUNNING, KafkaStreams.State.PENDING_SHUTDOWN),
+ KeyValue.pair(KafkaStreams.State.PENDING_SHUTDOWN, KafkaStreams.State.NOT_RUNNING)
+ )
+ );
+ private static final List> CRASH =
+ Collections.unmodifiableList(
+ Collections.singletonList(
+ KeyValue.pair(KafkaStreams.State.RUNNING, KafkaStreams.State.ERROR)
+ )
+ );
+ private static final List> CLOSE_CRASHED =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ KeyValue.pair(KafkaStreams.State.ERROR, KafkaStreams.State.PENDING_SHUTDOWN),
+ KeyValue.pair(KafkaStreams.State.PENDING_SHUTDOWN, KafkaStreams.State.NOT_RUNNING)
+ )
+ );
+
+ @ClassRule
+ public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(
+ NUM_BROKERS,
+ Utils.mkProperties(Collections.singletonMap("auto.create.topics.enable", "false"))
+ );
+
+ private static String applicationId;
+ private final static int NUM_TOPIC_PARTITIONS = 4;
+ private final static String CONSUMER_GROUP_ID = "readCommitted";
+ private final static String MULTI_PARTITION_INPUT_TOPIC = "multiPartitionInputTopic";
+ private final static String MULTI_PARTITION_OUTPUT_TOPIC = "multiPartitionOutputTopic";
+ private final String storeName = "store";
+
+ private final AtomicBoolean errorInjectedClient1 = new AtomicBoolean(false);
+ private final AtomicBoolean errorInjectedClient2 = new AtomicBoolean(false);
+ private final AtomicBoolean commitErrorInjectedClient1 = new AtomicBoolean(false);
+ private final AtomicBoolean commitErrorInjectedClient2 = new AtomicBoolean(false);
+ private final AtomicInteger commitCounterClient1 = new AtomicInteger(-1);
+ private final AtomicInteger commitCounterClient2 = new AtomicInteger(-1);
+ private final AtomicInteger commitRequested = new AtomicInteger(0);
+
+ private Throwable uncaughtException;
+
+ private int testNumber = 0;
+
+ @Before
+ public void createTopics() throws Exception {
+ applicationId = "appId-" + ++testNumber;
+ CLUSTER.deleteTopicsAndWait(
+ MULTI_PARTITION_INPUT_TOPIC,
+ MULTI_PARTITION_OUTPUT_TOPIC,
+ applicationId + "-" + storeName + "-changelog"
+ );
+
+ CLUSTER.createTopic(MULTI_PARTITION_INPUT_TOPIC, NUM_TOPIC_PARTITIONS, 1);
+ CLUSTER.createTopic(MULTI_PARTITION_OUTPUT_TOPIC, NUM_TOPIC_PARTITIONS, 1);
+ }
+
+ @Test
+ public void shouldUpgradeFromEosAlphaToEosBeta() throws Exception {
+ // We use two KafkaStreams clients that we upgrade from eos-alpha to eos-beta. During the upgrade,
+ // we ensure that there are pending transaction and verify that data is processed correctly.
+ //
+ // We either close clients cleanly (`injectError = false`) or let them crash (`injectError = true`) during
+ // the upgrade. For both cases, EOS should not be violated.
+ //
+ // Additionally, we inject errors while one client is on eos-alpha while the other client is on eos-beta:
+ // For this case, we inject the error during task commit phase, i.e., after offsets are appended to a TX,
+ // and before the TX is committed. The goal is to verify that the written but uncommitted offsets are not
+ // picked up, i.e., GroupCoordinator fencing works correctly.
+ //
+ // The commit interval is set to MAX_VALUE and the used `Processor` request commits manually so we have full
+ // control when a commit actually happens. We use an input topic with 4 partitions and each task will request
+ // a commit after processing 10 records.
+ //
+ // 1. start both clients and wait until rebalance stabilizes
+ // 2. write 10 records per input topic partition and verify that the result was committed
+ // 3. write 5 records per input topic partition to get pending transactions (verified via "read_uncommitted" mode)
+ // - all 4 pending transactions are based on task producers
+ // - we will get only 4 pending writes for one partition for the crash case as we crash processing the 5th record
+ // 4. stop/crash the first client, wait until rebalance stabilizes:
+ // - stop case:
+ // * verify that the stopped client did commit its pending transaction during shutdown
+ // * the second client will still have two pending transaction
+ // - crash case:
+ // * the pending transactions of the crashed client got aborted
+ // * the second client will have four pending transactions
+ // 5. restart the first client with eos-beta enabled and wait until rebalance stabilizes
+ // - the rebalance should result in a commit of all tasks
+ // 6. write 5 record per input topic partition
+ // - stop case:
+ // * verify that the result was committed
+ // - crash case:
+ // * fail the second (i.e., eos-alpha) client during commit
+ // * the eos-beta client should not pickup the pending offsets
+ // * verify uncommitted and committed result
+ // 7. only for crash case:
+ // 7a. restart the second client in eos-alpha mode and wait until rebalance stabilizes
+ // 7b. write 10 records per input topic partition
+ // * fail the first (i.e., eos-beta) client during commit
+ // * the eos-alpha client should not pickup the pending offsets
+ // * verify uncommitted and committed result
+ // 7c. restart the first client in eos-beta mode and wait until rebalance stabilizes
+ // 8. write 5 records per input topic partition to get pending transactions (verified via "read_uncommitted" mode)
+ // - 2 transaction are base on a task producer; one transaction is based on a thread producer
+ // - we will get 4 pending writes for the crash case as we crash processing the 5th record
+ // 9. stop/crash the second client and wait until rebalance stabilizes:
+ // - stop only:
+ // * verify that the stopped client did commit its pending transaction during shutdown
+ // * the first client will still have one pending transaction
+ // - crash case:
+ // * the pending transactions of the crashed client got aborted
+ // * the first client will have one pending transactions
+ // 10. restart the second client with eos-beta enabled and wait until rebalance stabilizes
+ // - the rebalance should result in a commit of all tasks
+ // 11. write 5 record per input topic partition and verify that the result was committed
+
+ final List> stateTransitions1 = new LinkedList<>();
+ KafkaStreams streams1Alpha = null;
+ KafkaStreams streams1Beta = null;
+ KafkaStreams streams1BetaTwo = null;
+
+ final List> stateTransitions2 = new LinkedList<>();
+ KafkaStreams streams2Alpha = null;
+ KafkaStreams streams2AlphaTwo = null;
+ KafkaStreams streams2Beta = null;
+
+ try {
+ // phase 1: start both clients
+ streams1Alpha = getKafkaStreams("appDir1", StreamsConfig.EXACTLY_ONCE);
+ streams1Alpha.setStateListener(
+ (newState, oldState) -> stateTransitions1.add(KeyValue.pair(oldState, newState))
+ );
+ streams1Alpha.cleanUp();
+ streams1Alpha.start();
+ waitForStateTransition(
+ stateTransitions1,
+ Arrays.asList(
+ KeyValue.pair(KafkaStreams.State.CREATED, KafkaStreams.State.REBALANCING),
+ KeyValue.pair(KafkaStreams.State.REBALANCING, KafkaStreams.State.RUNNING)
+ )
+ );
+
+ stateTransitions1.clear();
+ streams2Alpha = getKafkaStreams("appDir2", StreamsConfig.EXACTLY_ONCE);
+ streams2Alpha.setStateListener(
+ (newState, oldState) -> stateTransitions2.add(KeyValue.pair(oldState, newState))
+ );
+ streams2Alpha.cleanUp();
+ streams2Alpha.start();
+ waitForStateTransition(stateTransitions1, TWO_REBALANCES_RUNNING, stateTransitions2, TWO_REBALANCES_STARTUP);
+
+ // in all phases, we write comments that assume that p-0/p-1 are assigned to the first client
+ // and p-2/p-3 are assigned to the second client (in reality the assignment might be different though)
+
+ // phase 2: (write first batch of data)
+ // expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
+ //
+ // p-0: ---> 10 rec + C
+ // p-1: ---> 10 rec + C
+ // p-2: ---> 10 rec + C
+ // p-3: ---> 10 rec + C
+ final List> committedInputDataBeforeUpgrade =
+ prepareData(0L, 10L, 0L, 1L, 2L, 3L);
+ writeInputData(committedInputDataBeforeUpgrade);
+
+ waitForCondition(
+ () -> commitRequested.get() == 4,
+ MAX_WAIT_TIME_MS,
+ "SteamsTasks did not request commit."
+ );
+
+ final Map committedState = new HashMap<>();
+ final List> expectedUncommittedResult =
+ computeExpectedResult(committedInputDataBeforeUpgrade, committedState);
+ verifyCommitted(expectedUncommittedResult);
+
+ // phase 3: (write partial second batch of data)
+ // expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
+ //
+ // stop case:
+ // p-0: 10 rec + C ---> 5 rec (pending)
+ // p-1: 10 rec + C ---> 5 rec (pending)
+ // p-2: 10 rec + C ---> 5 rec (pending)
+ // p-3: 10 rec + C ---> 5 rec (pending)
+ // crash case: (we just assumes that we inject the error for p-0; in reality it might be a different partition)
+ // p-0: 10 rec + C ---> 4 rec (pending)
+ // p-1: 10 rec + C ---> 5 rec (pending)
+ // p-2: 10 rec + C ---> 5 rec (pending)
+ // p-3: 10 rec + C ---> 5 rec (pending)
+ final Set cleanKeys = mkSet(0L, 1L, 2L, 3L);
+ final Set keyFilterFirstClient = keysFromInstance(streams1Alpha);
+ final long potentiallyFirstFailingKey = keyFilterFirstClient.iterator().next();
+ cleanKeys.remove(potentiallyFirstFailingKey);
+
+ final List> uncommittedInputDataBeforeFirstUpgrade = new LinkedList<>();
+ if (!injectError) {
+ uncommittedInputDataBeforeFirstUpgrade.addAll(
+ prepareData(10L, 15L, 0L, 1L, 2L, 3L)
+ );
+ writeInputData(uncommittedInputDataBeforeFirstUpgrade);
+
+ expectedUncommittedResult.addAll(
+ computeExpectedResult(uncommittedInputDataBeforeFirstUpgrade, new HashMap<>(committedState))
+ );
+ verifyUncommitted(expectedUncommittedResult);
+ } else {
+ final List> uncommittedInputDataWithoutFailingKey = new LinkedList<>();
+ for (final long key : cleanKeys) {
+ uncommittedInputDataWithoutFailingKey.addAll(prepareData(10L, 15L, key));
+ }
+ uncommittedInputDataWithoutFailingKey.addAll(
+ prepareData(10L, 14L, potentiallyFirstFailingKey)
+ );
+ uncommittedInputDataBeforeFirstUpgrade.addAll(uncommittedInputDataWithoutFailingKey);
+ writeInputData(uncommittedInputDataWithoutFailingKey);
+
+ expectedUncommittedResult.addAll(
+ computeExpectedResult(uncommittedInputDataWithoutFailingKey, new HashMap<>(committedState))
+ );
+ verifyUncommitted(expectedUncommittedResult);
+ }
+
+ // phase 4: (stop/crash first client)
+ // expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
+ //
+ // stop case:
+ // p-0: 10 rec + C + 5 rec ---> C
+ // p-1: 10 rec + C + 5 rec ---> C
+ // p-2: 10 rec + C + 5 rec (pending)
+ // p-3: 10 rec + C + 5 rec (pending)
+ // crash case:
+ // p-0: 10 rec + C + 4 rec ---> A + 5 rec (pending)
+ // p-1: 10 rec + C + 5 rec ---> A + 5 rec (pending)
+ // p-2: 10 rec + C + 5 rec (pending)
+ // p-3: 10 rec + C + 5 rec (pending)
+ stateTransitions2.clear();
+ if (!injectError) {
+ stateTransitions1.clear();
+ streams1Alpha.close();
+ waitForStateTransition(stateTransitions1, CLOSE);
+ } else {
+ errorInjectedClient1.set(true);
+
+ final List> dataPotentiallyFirstFailingKey =
+ prepareData(14L, 15L, potentiallyFirstFailingKey);
+ uncommittedInputDataBeforeFirstUpgrade.addAll(dataPotentiallyFirstFailingKey);
+ writeInputData(dataPotentiallyFirstFailingKey);
+ }
+ waitForStateTransition(stateTransitions2, SINGLE_REBALANCE);
+
+ if (!injectError) {
+ final List> committedInputDataDuringFirstUpgrade =
+ uncommittedInputDataBeforeFirstUpgrade
+ .stream()
+ .filter(pair -> keyFilterFirstClient.contains(pair.key))
+ .collect(Collectors.toList());
+
+ final List> expectedCommittedResult =
+ computeExpectedResult(committedInputDataDuringFirstUpgrade, committedState);
+ verifyCommitted(expectedCommittedResult);
+ } else {
+ // retrying TX
+ expectedUncommittedResult.addAll(computeExpectedResult(
+ uncommittedInputDataBeforeFirstUpgrade
+ .stream()
+ .filter(pair -> keyFilterFirstClient.contains(pair.key))
+ .collect(Collectors.toList()),
+ new HashMap<>(committedState)
+ ));
+ verifyUncommitted(expectedUncommittedResult);
+
+ errorInjectedClient1.set(false);
+ stateTransitions1.clear();
+ streams1Alpha.close();
+ waitForStateTransition(stateTransitions1, CLOSE_CRASHED);
+ }
+
+ // phase 5: (restart first client)
+ // expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
+ //
+ // stop case:
+ // p-0: 10 rec + C + 5 rec + C
+ // p-1: 10 rec + C + 5 rec + C
+ // p-2: 10 rec + C + 5 rec ---> C
+ // p-3: 10 rec + C + 5 rec ---> C
+ // crash case:
+ // p-0: 10 rec + C + 4 rec + A + 5 rec ---> C
+ // p-1: 10 rec + C + 5 rec + A + 5 rec ---> C
+ // p-2: 10 rec + C + 5 rec ---> C
+ // p-3: 10 rec + C + 5 rec ---> C
+ commitRequested.set(0);
+ stateTransitions1.clear();
+ stateTransitions2.clear();
+ streams1Beta = getKafkaStreams("appDir1", StreamsConfig.EXACTLY_ONCE_BETA);
+ streams1Beta.setStateListener((newState, oldState) -> stateTransitions1.add(KeyValue.pair(oldState, newState)));
+ streams1Beta.start();
+ waitForStateTransition(stateTransitions1, TWO_REBALANCES_STARTUP, stateTransitions2, TWO_REBALANCES_RUNNING);
+
+ final Set committedKeys = mkSet(0L, 1L, 2L, 3L);
+ if (!injectError) {
+ committedKeys.removeAll(keyFilterFirstClient);
+ }
+
+ final List> expectedCommittedResultAfterRestartFirstClient = computeExpectedResult(
+ uncommittedInputDataBeforeFirstUpgrade
+ .stream()
+ .filter(pair -> committedKeys.contains(pair.key))
+ .collect(Collectors.toList()),
+ committedState
+ );
+ verifyCommitted(expectedCommittedResultAfterRestartFirstClient);
+
+ // phase 6: (complete second batch of data; crash: let second client fail on commit)
+ // expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
+ //
+ // stop case:
+ // p-0: 10 rec + C + 5 rec + C ---> 5 rec + C
+ // p-1: 10 rec + C + 5 rec + C ---> 5 rec + C
+ // p-2: 10 rec + C + 5 rec + C ---> 5 rec + C
+ // p-3: 10 rec + C + 5 rec + C ---> 5 rec + C
+ // crash case:
+ // p-0: 10 rec + C + 4 rec + A + 5 rec + C ---> 5 rec + C
+ // p-1: 10 rec + C + 5 rec + A + 5 rec + C ---> 5 rec + C
+ // p-2: 10 rec + C + 5 rec + C ---> 5 rec + A + 5 rec + C
+ // p-3: 10 rec + C + 5 rec + C ---> 5 rec + A + 5 rec + C
+ commitCounterClient1.set(0);
+
+ if (!injectError) {
+ final List> committedInputDataDuringUpgrade =
+ prepareData(15L, 20L, 0L, 1L, 2L, 3L);
+ writeInputData(committedInputDataDuringUpgrade);
+
+ final List> expectedCommittedResult =
+ computeExpectedResult(committedInputDataDuringUpgrade, committedState);
+ verifyCommitted(expectedCommittedResult);
+ expectedUncommittedResult.addAll(expectedCommittedResult);
+ } else {
+ final Set keysFirstClient = keysFromInstance(streams1Beta);
+ final Set keysSecondClient = keysFromInstance(streams2Alpha);
+
+ final List> committedInputDataAfterFirstUpgrade =
+ prepareData(15L, 20L, keysFirstClient.toArray(new Long[0]));
+ writeInputData(committedInputDataAfterFirstUpgrade);
+
+ final List> expectedCommittedResultBeforeFailure =
+ computeExpectedResult(committedInputDataAfterFirstUpgrade, committedState);
+ verifyCommitted(expectedCommittedResultBeforeFailure);
+ expectedUncommittedResult.addAll(expectedCommittedResultBeforeFailure);
+
+ commitCounterClient2.set(0);
+
+ final Iterator it = keysSecondClient.iterator();
+ final Long otherKey = it.next();
+ final Long failingKey = it.next();
+
+ final List> uncommittedInputDataAfterFirstUpgrade =
+ prepareData(15L, 19L, keysSecondClient.toArray(new Long[0]));
+ uncommittedInputDataAfterFirstUpgrade.addAll(prepareData(19L, 20L, otherKey));
+ writeInputData(uncommittedInputDataAfterFirstUpgrade);
+
+ final Map uncommittedState = new HashMap<>(committedState);
+ expectedUncommittedResult.addAll(
+ computeExpectedResult(uncommittedInputDataAfterFirstUpgrade, uncommittedState)
+ );
+ verifyUncommitted(expectedUncommittedResult);
+
+ stateTransitions1.clear();
+ stateTransitions2.clear();
+ commitCounterClient1.set(0);
+ commitErrorInjectedClient2.set(true);
+
+ final List> dataFailingKey = prepareData(19L, 20L, failingKey);
+ uncommittedInputDataAfterFirstUpgrade.addAll(dataFailingKey);
+ writeInputData(dataFailingKey);
+
+ expectedUncommittedResult.addAll(
+ computeExpectedResult(dataFailingKey, uncommittedState)
+ );
+ verifyUncommitted(expectedUncommittedResult);
+
+ waitForStateTransition(stateTransitions1, SINGLE_REBALANCE, stateTransitions2, CRASH);
+
+ commitErrorInjectedClient2.set(false);
+ stateTransitions2.clear();
+ streams2Alpha.close();
+ waitForStateTransition(stateTransitions2, CLOSE_CRASHED);
+
+ final List> expectedCommittedResultAfterFailure =
+ computeExpectedResult(uncommittedInputDataAfterFirstUpgrade, committedState);
+ verifyCommitted(expectedCommittedResultAfterFailure);
+ expectedUncommittedResult.addAll(expectedCommittedResultAfterFailure);
+ }
+
+ // 7. only for crash case:
+ // 7a. restart the second client in eos-alpha mode and wait until rebalance stabilizes
+ // 7b. write third batch of input data
+ // * fail the first (i.e., eos-beta) client during commit
+ // * the eos-alpha client should not pickup the pending offsets
+ // * verify uncommitted and committed result
+ // 7c. restart the first client in eos-beta mode and wait until rebalance stabilizes
+ //
+ // crash case:
+ // p-0: 10 rec + C + 4 rec + A + 5 rec + C + 5 rec + C ---> 10 rec + A + 10 rec + C
+ // p-1: 10 rec + C + 5 rec + A + 5 rec + C + 5 rec + C ---> 10 rec + A + 10 rec + C
+ // p-2: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C ---> 10 rec + C
+ // p-3: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C ---> 10 rec + C
+ if (!injectError) {
+ streams2AlphaTwo = streams2Alpha;
+ } else {
+ // 7a restart the second client in eos-alpha mode and wait until rebalance stabilizes
+ commitCounterClient1.set(0);
+ commitCounterClient2.set(-1);
+ stateTransitions1.clear();
+ stateTransitions2.clear();
+ streams2AlphaTwo = getKafkaStreams("appDir2", StreamsConfig.EXACTLY_ONCE);
+ streams2AlphaTwo.setStateListener(
+ (newState, oldState) -> stateTransitions2.add(KeyValue.pair(oldState, newState))
+ );
+ streams2AlphaTwo.start();
+ waitForStateTransition(stateTransitions1, TWO_REBALANCES_RUNNING, stateTransitions2, TWO_REBALANCES_STARTUP);
+
+ // 7b. write third batch of input data
+ final Set keysFirstClient = keysFromInstance(streams1Beta);
+ final Set keysSecondClient = keysFromInstance(streams2AlphaTwo);
+
+ final List> committedInputDataBetweenUpgrades =
+ prepareData(20L, 30L, keysSecondClient.toArray(new Long[0]));
+ writeInputData(committedInputDataBetweenUpgrades);
+
+ final List> expectedCommittedResultBeforeFailure =
+ computeExpectedResult(committedInputDataBetweenUpgrades, committedState);
+ verifyCommitted(expectedCommittedResultBeforeFailure);
+ expectedUncommittedResult.addAll(expectedCommittedResultBeforeFailure);
+
+ commitCounterClient2.set(0);
+
+ final Iterator it = keysFirstClient.iterator();
+ final Long otherKey = it.next();
+ final Long failingKey = it.next();
+
+ final List> uncommittedInputDataBetweenUpgrade =
+ prepareData(20L, 29L, keysFirstClient.toArray(new Long[0]));
+ uncommittedInputDataBetweenUpgrade.addAll(prepareData(29L, 30L, otherKey));
+ writeInputData(uncommittedInputDataBetweenUpgrade);
+
+ final Map uncommittedState = new HashMap<>(committedState);
+ expectedUncommittedResult.addAll(
+ computeExpectedResult(uncommittedInputDataBetweenUpgrade, uncommittedState)
+ );
+ verifyUncommitted(expectedUncommittedResult);
+
+ stateTransitions1.clear();
+ stateTransitions2.clear();
+ commitCounterClient2.set(0);
+ commitErrorInjectedClient1.set(true);
+
+ final List> dataFailingKey = prepareData(29L, 30L, failingKey);
+ uncommittedInputDataBetweenUpgrade.addAll(dataFailingKey);
+ writeInputData(dataFailingKey);
+
+ expectedUncommittedResult.addAll(
+ computeExpectedResult(dataFailingKey, uncommittedState)
+ );
+ verifyUncommitted(expectedUncommittedResult);
+
+ waitForStateTransition(stateTransitions1, CRASH, stateTransitions2, SINGLE_REBALANCE);
+
+ commitErrorInjectedClient1.set(false);
+ stateTransitions1.clear();
+ streams1Beta.close();
+ waitForStateTransition(stateTransitions1, CLOSE_CRASHED);
+
+ final List> expectedCommittedResultAfterFailure =
+ computeExpectedResult(uncommittedInputDataBetweenUpgrade, committedState);
+ verifyCommitted(expectedCommittedResultAfterFailure);
+ expectedUncommittedResult.addAll(expectedCommittedResultAfterFailure);
+
+ // 7c. restart the first client in eos-beta mode and wait until rebalance stabilizes
+ stateTransitions1.clear();
+ stateTransitions2.clear();
+ streams1BetaTwo = getKafkaStreams("appDir1", StreamsConfig.EXACTLY_ONCE_BETA);
+ streams1BetaTwo.setStateListener((newState, oldState) -> stateTransitions1.add(KeyValue.pair(oldState, newState)));
+ streams1BetaTwo.start();
+ waitForStateTransition(stateTransitions1, TWO_REBALANCES_STARTUP, stateTransitions2, TWO_REBALANCES_RUNNING);
+ }
+
+ // phase 8: (write partial fourth batch of data)
+ // expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
+ //
+ // stop case:
+ // p-0: 10 rec + C + 5 rec + C + 5 rec + C ---> 5 rec (pending)
+ // p-1: 10 rec + C + 5 rec + C + 5 rec + C ---> 5 rec (pending)
+ // p-2: 10 rec + C + 5 rec + C + 5 rec + C ---> 5 rec (pending)
+ // p-3: 10 rec + C + 5 rec + C + 5 rec + C ---> 5 rec (pending)
+ // crash case: (we just assumes that we inject the error for p-2; in reality it might be a different partition)
+ // p-0: 10 rec + C + 4 rec + A + 5 rec + C + 5 rec + C + 10 rec + A + 10 rec + C ---> 5 rec (pending)
+ // p-1: 10 rec + C + 5 rec + A + 5 rec + C + 5 rec + C + 10 rec + A + 10 rec + C ---> 5 rec (pending)
+ // p-2: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C + 10 rec + C ---> 4 rec (pending)
+ // p-3: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C + 10 rec + C ---> 5 rec (pending)
+ cleanKeys.addAll(mkSet(0L, 1L, 2L, 3L));
+ final Set keyFilterSecondClient = keysFromInstance(streams2AlphaTwo);
+ final long potentiallySecondFailingKey = keyFilterSecondClient.iterator().next();
+ cleanKeys.remove(potentiallySecondFailingKey);
+
+ final List> uncommittedInputDataBeforeSecondUpgrade = new LinkedList<>();
+ if (!injectError) {
+ uncommittedInputDataBeforeSecondUpgrade.addAll(
+ prepareData(30L, 35L, 0L, 1L, 2L, 3L)
+ );
+ writeInputData(uncommittedInputDataBeforeSecondUpgrade);
+
+ expectedUncommittedResult.addAll(
+ computeExpectedResult(uncommittedInputDataBeforeSecondUpgrade, new HashMap<>(committedState))
+ );
+ verifyUncommitted(expectedUncommittedResult);
+ } else {
+ final List> uncommittedInputDataWithoutFailingKey = new LinkedList<>();
+ for (final long key : cleanKeys) {
+ uncommittedInputDataWithoutFailingKey.addAll(prepareData(30L, 35L, key));
+ }
+ uncommittedInputDataWithoutFailingKey.addAll(
+ prepareData(30L, 34L, potentiallySecondFailingKey)
+ );
+ uncommittedInputDataBeforeSecondUpgrade.addAll(uncommittedInputDataWithoutFailingKey);
+ writeInputData(uncommittedInputDataWithoutFailingKey);
+
+ expectedUncommittedResult.addAll(
+ computeExpectedResult(uncommittedInputDataWithoutFailingKey, new HashMap<>(committedState))
+ );
+ verifyUncommitted(expectedUncommittedResult);
+ }
+
+ // phase 9: (stop/crash second client)
+ // expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
+ //
+ // stop case:
+ // p-0: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec (pending)
+ // p-1: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec (pending)
+ // p-2: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec ---> C
+ // p-3: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec ---> C
+ // crash case: (we just assumes that we inject the error for p-2; in reality it might be a different partition)
+ // p-0: 10 rec + C + 4 rec + A + 5 rec + C + 5 rec + C + 10 rec + A + 10 rec + C + 5 rec (pending)
+ // p-1: 10 rec + C + 5 rec + A + 5 rec + C + 5 rec + C + 10 rec + A + 10 rec + C + 5 rec (pending)
+ // p-2: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C + 10 rec + C + 4 rec ---> A + 5 rec (pending)
+ // p-3: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C + 10 rec + C + 5 rec ---> A + 5 rec (pending)
+ stateTransitions1.clear();
+ if (!injectError) {
+ stateTransitions2.clear();
+ streams2AlphaTwo.close();
+ waitForStateTransition(stateTransitions2, CLOSE);
+ } else {
+ errorInjectedClient2.set(true);
+
+ final List> dataPotentiallySecondFailingKey =
+ prepareData(34L, 35L, potentiallySecondFailingKey);
+ uncommittedInputDataBeforeSecondUpgrade.addAll(dataPotentiallySecondFailingKey);
+ writeInputData(dataPotentiallySecondFailingKey);
+ }
+ waitForStateTransition(stateTransitions1, SINGLE_REBALANCE);
+
+ if (!injectError) {
+ final List> committedInputDataDuringSecondUpgrade =
+ uncommittedInputDataBeforeSecondUpgrade
+ .stream()
+ .filter(pair -> keyFilterSecondClient.contains(pair.key))
+ .collect(Collectors.toList());
+
+ final List> expectedCommittedResult =
+ computeExpectedResult(committedInputDataDuringSecondUpgrade, committedState);
+ verifyCommitted(expectedCommittedResult);
+ } else {
+ // retrying TX
+ expectedUncommittedResult.addAll(computeExpectedResult(
+ uncommittedInputDataBeforeSecondUpgrade
+ .stream()
+ .filter(pair -> keyFilterSecondClient.contains(pair.key))
+ .collect(Collectors.toList()),
+ new HashMap<>(committedState)
+ ));
+ verifyUncommitted(expectedUncommittedResult);
+
+ errorInjectedClient2.set(false);
+ stateTransitions2.clear();
+ streams2AlphaTwo.close();
+ waitForStateTransition(stateTransitions2, CLOSE_CRASHED);
+ }
+
+ // phase 10: (restart second client)
+ // expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
+ //
+ // the state below indicate the case for which the "original" tasks of client2 are migrated back to client2
+ // if a task "switch" happens, we might get additional commits (omitted in the comment for brevity)
+ //
+ // stop case:
+ // p-0: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec ---> C
+ // p-1: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec ---> C
+ // p-2: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec + C
+ // p-3: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec + C
+ // crash case: (we just assumes that we inject the error for p-2; in reality it might be a different partition)
+ // p-0: 10 rec + C + 4 rec + A + 5 rec + C + 5 rec + C + 10 rec + A + 10 rec + C + 5 rec ---> C
+ // p-1: 10 rec + C + 5 rec + A + 5 rec + C + 5 rec + C + 10 rec + A + 10 rec + C + 5 rec ---> C
+ // p-2: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C + 10 rec + C + 4 rec + A + 5 rec ---> C
+ // p-3: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C + 10 rec + C + 5 rec + A + 5 rec ---> C
+ commitRequested.set(0);
+ stateTransitions1.clear();
+ stateTransitions2.clear();
+ streams2Beta = getKafkaStreams("appDir1", StreamsConfig.EXACTLY_ONCE_BETA);
+ streams2Beta.setStateListener(
+ (newState, oldState) -> stateTransitions2.add(KeyValue.pair(oldState, newState))
+ );
+ streams2Beta.start();
+ waitForStateTransition(stateTransitions1, TWO_REBALANCES_RUNNING, stateTransitions2, TWO_REBALANCES_STARTUP);
+
+ committedKeys.addAll(mkSet(0L, 1L, 2L, 3L));
+ if (!injectError) {
+ committedKeys.removeAll(keyFilterSecondClient);
+ }
+
+ final List> expectedCommittedResultAfterRestartSecondClient = computeExpectedResult(
+ uncommittedInputDataBeforeSecondUpgrade
+ .stream()
+ .filter(pair -> committedKeys.contains(pair.key))
+ .collect(Collectors.toList()),
+ committedState
+ );
+ verifyCommitted(expectedCommittedResultAfterRestartSecondClient);
+
+ // phase 11: (complete fourth batch of data)
+ // expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
+ //
+ // stop case:
+ // p-0: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec + C ---> 5 rec + C
+ // p-1: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec + C ---> 5 rec + C
+ // p-2: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec + C ---> 5 rec + C
+ // p-3: 10 rec + C + 5 rec + C + 5 rec + C + 5 rec + C ---> 5 rec + C
+ // crash case: (we just assumes that we inject the error for p-2; in reality it might be a different partition)
+ // p-0: 10 rec + C + 4 rec + A + 5 rec + C + 5 rec + C + 10 rec + A + 10 rec + C + 5 rec + C ---> 5 rec + C
+ // p-1: 10 rec + C + 5 rec + A + 5 rec + C + 5 rec + C + 10 rec + A + 10 rec + C + 5 rec + C ---> 5 rec + C
+ // p-2: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C + 10 rec + C + 4 rec + A + 5 rec + C ---> 5 rec + C
+ // p-3: 10 rec + C + 5 rec + C + 5 rec + A + 5 rec + C + 10 rec + C + 5 rec + A + 5 rec + C ---> 5 rec + C
+ commitCounterClient1.set(-1);
+ commitCounterClient2.set(-1);
+
+ final List> committedInputDataAfterUpgrade =
+ prepareData(35L, 40L, 0L, 1L, 2L, 3L);
+ writeInputData(committedInputDataAfterUpgrade);
+
+ final List> expectedCommittedResult =
+ computeExpectedResult(committedInputDataAfterUpgrade, committedState);
+ verifyCommitted(expectedCommittedResult);
+ } finally {
+ if (streams1Alpha != null) {
+ streams1Alpha.close();
+ }
+ if (streams1Beta != null) {
+ streams1Beta.close();
+ }
+ if (streams1BetaTwo != null) {
+ streams1BetaTwo.close();
+ }
+ if (streams2Alpha != null) {
+ streams2Alpha.close();
+ }
+ if (streams2AlphaTwo != null) {
+ streams2AlphaTwo.close();
+ }
+ if (streams2Beta != null) {
+ streams2Beta.close();
+ }
+ }
+ }
+
+ private KafkaStreams getKafkaStreams(final String appDir,
+ final String processingGuarantee) {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final String[] storeNames = new String[] {storeName};
+ final StoreBuilder> storeBuilder = Stores
+ .keyValueStoreBuilder(Stores.persistentKeyValueStore(storeName), Serdes.Long(), Serdes.Long())
+ .withCachingEnabled();
+
+ builder.addStateStore(storeBuilder);
+
+ final KStream input = builder.stream(MULTI_PARTITION_INPUT_TOPIC);
+ input.transform(new TransformerSupplier>() {
+ @SuppressWarnings("unchecked")
+ @Override
+ public Transformer> get() {
+ return new Transformer>() {
+ ProcessorContext context;
+ KeyValueStore state = null;
+ AtomicBoolean crash;
+ AtomicInteger sharedCommit;
+
+ @Override
+ public void init(final ProcessorContext context) {
+ this.context = context;
+ state = (KeyValueStore) context.getStateStore(storeName);
+ final String clientId = context.appConfigs().get(StreamsConfig.CLIENT_ID_CONFIG).toString();
+ if ("appDir1".equals(clientId)) {
+ crash = errorInjectedClient1;
+ sharedCommit = commitCounterClient1;
+ } else {
+ crash = errorInjectedClient2;
+ sharedCommit = commitCounterClient2;
+ }
+ }
+
+ @Override
+ public KeyValue transform(final Long key, final Long value) {
+ if ((value + 1) % 10 == 0) {
+ if (sharedCommit.get() < 0 ||
+ sharedCommit.incrementAndGet() == 2) {
+
+ context.commit();
+ }
+ commitRequested.incrementAndGet();
+ }
+
+ Long sum = state.get(key);
+ if (sum == null) {
+ sum = value;
+ } else {
+ sum += value;
+ }
+ state.put(key, sum);
+ state.flush();
+
+ if (value % 10 == 4 && // potentially crash when processing 5th, 15th, or 25th record (etc.)
+ crash != null && crash.compareAndSet(true, false)) {
+ // only crash a single task
+ throw new RuntimeException("Injected test exception.");
+ }
+
+ return new KeyValue<>(key, state.get(key));
+ }
+
+ @Override
+ public void close() { }
+ };
+ } }, storeNames)
+ .to(MULTI_PARTITION_OUTPUT_TOPIC);
+
+ final Properties properties = new Properties();
+ properties.put(StreamsConfig.CLIENT_ID_CONFIG, appDir);
+ properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuarantee);
+ properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Long.MAX_VALUE);
+ properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.METADATA_MAX_AGE_CONFIG), "1000");
+ properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), "earliest");
+ properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG), 5 * 1000);
+ properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG), 5 * 1000 - 1);
+ properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG), MAX_POLL_INTERVAL_MS);
+ properties.put(StreamsConfig.producerPrefix(ProducerConfig.PARTITIONER_CLASS_CONFIG), KeyPartitioner.class);
+ properties.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
+ properties.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath() + File.separator + appDir);
+
+ final Properties config = StreamsTestUtils.getStreamsConfig(
+ applicationId,
+ CLUSTER.bootstrapServers(),
+ Serdes.LongSerde.class.getName(),
+ Serdes.LongSerde.class.getName(),
+ properties
+ );
+
+ final KafkaStreams streams = new KafkaStreams(builder.build(), config, new TestKafkaClientSupplier());
+
+ streams.setUncaughtExceptionHandler((t, e) -> {
+ if (uncaughtException != null) {
+ e.printStackTrace(System.err);
+ fail("Should only get one uncaught exception from Streams.");
+ }
+ uncaughtException = e;
+ });
+
+ return streams;
+ }
+
+ private void waitForStateTransition(final List> observed,
+ final List> expected)
+ throws Exception {
+
+ waitForCondition(
+ () -> observed.equals(expected),
+ MAX_WAIT_TIME_MS,
+ () -> "Client did not startup on time. Observers transitions: " + observed
+ );
+ }
+
+ private void waitForStateTransition(final List> observed1,
+ final List> expected1,
+ final List> observed2,
+ final List> expected2)
+ throws Exception {
+
+ waitForCondition(
+ () -> observed1.equals(expected1) && observed2.equals(expected2),
+ MAX_WAIT_TIME_MS,
+ () -> "Clients did not startup and stabilize on time. Observed transitions: " +
+ "\n client-1 transitions: " + observed1 +
+ "\n client-2 transitions: " + observed2
+ );
+ }
+
+ private List> prepareData(final long fromInclusive,
+ final long toExclusive,
+ final Long... keys) {
+ final List> data = new ArrayList<>();
+
+ for (final Long k : keys) {
+ for (long v = fromInclusive; v < toExclusive; ++v) {
+ data.add(new KeyValue<>(k, v));
+ }
+ }
+
+ return data;
+ }
+
+ private void writeInputData(final List> records) {
+ final Properties config = TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ LongSerializer.class,
+ LongSerializer.class
+ );
+ config.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, KeyPartitioner.class.getName());
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ MULTI_PARTITION_INPUT_TOPIC,
+ records,
+ config,
+ CLUSTER.time
+ );
+ }
+
+ private void verifyCommitted(final List> expectedResult) throws Exception {
+ final List> committedOutput = readResult(expectedResult.size(), true);
+ checkResultPerKey(committedOutput, expectedResult);
+ }
+
+ private void verifyUncommitted(final List> expectedResult) throws Exception {
+ final List> uncommittedOutput = readResult(expectedResult.size(), false);
+ checkResultPerKey(uncommittedOutput, expectedResult);
+ }
+
+ private List> readResult(final int numberOfRecords,
+ final boolean readCommitted) throws Exception {
+ if (readCommitted) {
+ return IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
+ TestUtils.consumerConfig(
+ CLUSTER.bootstrapServers(),
+ CONSUMER_GROUP_ID,
+ LongDeserializer.class,
+ LongDeserializer.class,
+ Utils.mkProperties(Collections.singletonMap(
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG,
+ IsolationLevel.READ_COMMITTED.name().toLowerCase(Locale.ROOT))
+ )
+ ),
+ MULTI_PARTITION_OUTPUT_TOPIC,
+ numberOfRecords
+ );
+ }
+
+ // read uncommitted
+ return IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
+ TestUtils.consumerConfig(CLUSTER.bootstrapServers(), LongDeserializer.class, LongDeserializer.class),
+ MULTI_PARTITION_OUTPUT_TOPIC,
+ numberOfRecords
+ );
+ }
+
+ private void checkResultPerKey(final List> result,
+ final List> expectedResult) {
+ final Set allKeys = new HashSet<>();
+ addAllKeys(allKeys, result);
+ addAllKeys(allKeys, expectedResult);
+
+ for (final Long key : allKeys) {
+ assertThat(getAllRecordPerKey(key, result), equalTo(getAllRecordPerKey(key, expectedResult)));
+ }
+ }
+
+ private void addAllKeys(final Set allKeys, final List> records) {
+ for (final KeyValue record : records) {
+ allKeys.add(record.key);
+ }
+ }
+
+ private List> getAllRecordPerKey(final Long key, final List> records) {
+ final List> recordsPerKey = new ArrayList<>(records.size());
+
+ for (final KeyValue record : records) {
+ if (record.key.equals(key)) {
+ recordsPerKey.add(record);
+ }
+ }
+
+ return recordsPerKey;
+ }
+
+ private List> computeExpectedResult(final List> input,
+ final Map currentState) {
+ final List> expectedResult = new ArrayList<>(input.size());
+
+ for (final KeyValue record : input) {
+ final long sum = currentState.getOrDefault(record.key, 0L);
+ currentState.put(record.key, sum + record.value);
+ expectedResult.add(new KeyValue<>(record.key, sum + record.value));
+ }
+
+ return expectedResult;
+ }
+
+ private Set keysFromInstance(final KafkaStreams streams) {
+ final ReadOnlyKeyValueStore store =
+ streams.store(StoreQueryParameters.fromNameAndType(storeName, QueryableStoreTypes.keyValueStore()));
+ final Set keys = new HashSet<>();
+ try (final KeyValueIterator it = store.all()) {
+ while (it.hasNext()) {
+ final KeyValue row = it.next();
+ keys.add(row.key);
+ }
+ }
+ return keys;
+ }
+
+ // must be public to allow KafkaProducer to instantiate it
+ public static class KeyPartitioner implements Partitioner {
+ private final static LongDeserializer LONG_DESERIALIZER = new LongDeserializer();
+
+ @Override
+ public int partition(final String topic,
+ final Object key,
+ final byte[] keyBytes,
+ final Object value,
+ final byte[] valueBytes,
+ final Cluster cluster) {
+ return LONG_DESERIALIZER.deserialize(topic, keyBytes).intValue() % NUM_TOPIC_PARTITIONS;
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void configure(final Map configs) {}
+ }
+
+ private class TestKafkaClientSupplier extends DefaultKafkaClientSupplier {
+ @Override
+ public Producer getProducer(final Map config) {
+ return new ErrorInjector(config);
+ }
+ }
+
+ private class ErrorInjector extends KafkaProducer {
+ private final AtomicBoolean crash;
+
+ public ErrorInjector(final Map configs) {
+ super(configs, new ByteArraySerializer(), new ByteArraySerializer());
+ final String clientId = configs.get(ProducerConfig.CLIENT_ID_CONFIG).toString();
+ if (clientId.contains("appDir1")) {
+ crash = commitErrorInjectedClient1;
+ } else {
+ crash = commitErrorInjectedClient2;
+ }
+ }
+
+ @Override
+ public void commitTransaction() throws ProducerFencedException {
+ super.flush(); // we flush to ensure that the offsets are written
+ if (!crash.compareAndSet(true, false)) {
+ super.commitTransaction();
+ } else {
+ throw new RuntimeException("Injected producer commit exception.");
+ }
+ }
+ }
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
index fa65766d962ad..376282b19a1d9 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
@@ -279,7 +279,6 @@ private void checkResultPerKey(final List> result, final Li
for (final Long key : allKeys) {
assertThat(getAllRecordPerKey(key, result), equalTo(getAllRecordPerKey(key, expectedResult)));
}
-
}
private void addAllKeys(final Set allKeys, final List> records) {