Skip to content

KAFKA-9891: add integration tests for EOS and StandbyTask #8890

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guozhangwang merged 1 commit into from
Jun 19, 2020
Merged

KAFKA-9891: add integration tests for EOS and StandbyTask #8890

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/StoreQueryParameters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ private StoreQueryParameters(final String storeName, final QueryableStoreType<T>
}

public static <T> StoreQueryParameters<T> fromNameAndType(final String storeName,
final QueryableStoreType<T> queryableStoreType) {
final QueryableStoreType<T> queryableStoreType) {
return new StoreQueryParameters<T>(storeName, queryableStoreType, null, false);
}

Expand Down
230 changes: 223 additions & 7 deletions ...ams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

/**
Expand All @@ -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<String[]> data() {
return asList(new String[][] {
Expand All @@ -72,8 +84,12 @@ public static Collection<String[]> 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);
Expand All @@ -86,30 +102,36 @@ 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
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();

final CountDownLatch instanceLatch = new CountDownLatch(1);

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));

Expand All @@ -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);
Expand All @@ -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.<Integer, Integer>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.<Integer, Integer>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"

@vvcephei vvcephei Jun 19, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Did you mean "stale"?

startApplicationAndWaitUntilRunning(
Collections.singletonList(streamInstanceOneRecovery),
Duration.ofSeconds(30)
);
waitForCondition(
() -> streamInstanceOneRecovery.store(
StoreQueryParameters.fromNameAndType(
storeName,
QueryableStoreTypes.<Integer, Integer>keyValueStore()
).enableStaleStores()
).get(KEY_0) != null,
"Could not get key from recovered standby store"
);

streamInstanceTwo.close();
waitForCondition(
() -> streamInstanceOneRecovery.store(
StoreQueryParameters.fromNameAndType(
storeName,
QueryableStoreTypes.<Integer, Integer>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.<Integer, Integer>stream(inputTopic)
.transform(
() -> new Transformer<Integer, Integer, KeyValue<Integer, Integer>>() {
private KeyValueStore<Integer, Integer> store;

@SuppressWarnings("unchecked")
@Override
public void init(final ProcessorContext context) {
store = (KeyValueStore<Integer, Integer>) context.getStateStore(storeName);
}

@Override
public KeyValue<Integer, Integer> 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);
Expand All @@ -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);
Comment on lines +379 to +380

@vvcephei vvcephei Jun 19, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we really need this? It seems like the only thing that depends on knowing which instance would get the active is just waiting for the crash after the poison pill. Could we instead just wait for once of the instances to crash, but not worry about which?

@mjsax mjsax Jun 19, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. It's for the first phase of the test. We start the first instance and let it process the first record. As there is not enough capacity, no standby is scheduled. When we start the second instance, with "lag=0" setting, we ensure that the standby is placed at instance two. With default setting, we don't know which instance will get the active/standby assigned. -> when we inject the poison pill, we know that instance one will fail as it hosts the active.

@vvcephei vvcephei Jun 19, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right, this was my suggestion:

Could we instead just wait for once of the instances to crash, but not worry about which?

@mjsax mjsax Jun 19, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, it make the test more complex, because all the following code depends on instance one failing.

@vvcephei vvcephei Jun 19, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, it's your call. I think this might make the tests flaky, but I guess we can figure that out later.

streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

return streamsConfiguration;
Expand Down