Skip to content

KAFKA-8011: Fix flaky RegexSourceIntegrationTest #8799

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
mjsax merged 3 commits into from
Jun 5, 2020
Merged

KAFKA-8011: Fix flaky RegexSourceIntegrationTest #8799

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
897c200
KAFKA-8011: Fix flaky RegexSourceIntegrationTest
mjsax Jun 4, 2020
8c2a6d5
Minor
mjsax Jun 4, 2020
93f6343
Github comments
mjsax Jun 5, 2020
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
4 changes: 3 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,9 @@ ConsumerRecord<Object, Object> deserialize(final ProcessorContext processorConte
rawRecord.serializedKeySize(),
rawRecord.serializedValueSize(),
sourceNode.deserializeKey(rawRecord.topic(), rawRecord.headers(), rawRecord.key()),
sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value()), rawRecord.headers());
sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value()),
rawRecord.headers()
);
} catch (final Exception deserializationException) {
final DeserializationExceptionHandler.DeserializationHandlerResponse response;
try {
Expand Down
185 changes: 92 additions & 93 deletions streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,8 @@
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.test.IntegrationTest;
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.MockKeyValueStoreBuilder;
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestCondition;
import org.apache.kafka.test.TestUtils;
Expand All @@ -52,6 +52,7 @@
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TestName;

import java.io.IOException;
import java.util.ArrayList;
Expand Down Expand Up @@ -124,11 +125,13 @@ public void setUp() throws InterruptedException {
properties.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "1000");
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

streamsConfiguration = StreamsTestUtils.getStreamsConfig("regex-source-integration-test",
CLUSTER.bootstrapServers(),
STRING_SERDE_CLASSNAME,
STRING_SERDE_CLASSNAME,
properties);
streamsConfiguration = StreamsTestUtils.getStreamsConfig(
IntegrationTestUtils.safeUniqueTestName(RegexSourceIntegrationTest.class, new TestName()),
CLUSTER.bootstrapServers(),
STRING_SERDE_CLASSNAME,
STRING_SERDE_CLASSNAME,
properties
);
}

@After
Expand All @@ -142,83 +145,89 @@ public void tearDown() throws IOException {

@Test
public void testRegexMatchesTopicsAWhenCreated() throws Exception {
try {
final Serde<String> stringSerde = Serdes.String();

final Serde<String> stringSerde = Serdes.String();

final List<String> expectedFirstAssignment = Collections.singletonList("TEST-TOPIC-1");
// we compare lists of subscribed topics and hence requiring the order as well; this is guaranteed
// with KIP-429 since we would NOT revoke TEST-TOPIC-1 but only add TEST-TOPIC-2 so the list is always
// in the order of "TEST-TOPIC-1, TEST-TOPIC-2". Note if KIP-429 behavior ever changed it may become a flaky test
final List<String> expectedSecondAssignment = Arrays.asList("TEST-TOPIC-1", "TEST-TOPIC-2");
final List<String> expectedFirstAssignment = Collections.singletonList("TEST-TOPIC-1");
// we compare lists of subscribed topics and hence requiring the order as well; this is guaranteed
// with KIP-429 since we would NOT revoke TEST-TOPIC-1 but only add TEST-TOPIC-2 so the list is always
// in the order of "TEST-TOPIC-1, TEST-TOPIC-2". Note if KIP-429 behavior ever changed it may become a flaky test
final List<String> expectedSecondAssignment = Arrays.asList("TEST-TOPIC-1", "TEST-TOPIC-2");

CLUSTER.createTopic("TEST-TOPIC-1");
CLUSTER.createTopic("TEST-TOPIC-1");

final StreamsBuilder builder = new StreamsBuilder();
final StreamsBuilder builder = new StreamsBuilder();

final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-\\d"));
final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-\\d"));

pattern1Stream.to(outputTopic, Produced.with(stringSerde, stringSerde));
final List<String> assignedTopics = new CopyOnWriteArrayList<>();
streams = new KafkaStreams(builder.build(), streamsConfiguration, new DefaultKafkaClientSupplier() {
@Override
public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
return new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
@Override
public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {
super.subscribe(topics, new TheConsumerRebalanceListener(assignedTopics, listener));
}
};
pattern1Stream.to(outputTopic, Produced.with(stringSerde, stringSerde));
final List<String> assignedTopics = new CopyOnWriteArrayList<>();
streams = new KafkaStreams(builder.build(), streamsConfiguration, new DefaultKafkaClientSupplier() {
@Override
public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
return new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
@Override
public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {
super.subscribe(topics, new TheConsumerRebalanceListener(assignedTopics, listener));
}
};

}
});
}
});

streams.start();
TestUtils.waitForCondition(() -> assignedTopics.equals(expectedFirstAssignment), STREAM_TASKS_NOT_UPDATED);
streams.start();
TestUtils.waitForCondition(() -> assignedTopics.equals(expectedFirstAssignment), STREAM_TASKS_NOT_UPDATED);

CLUSTER.createTopic("TEST-TOPIC-2");
CLUSTER.createTopic("TEST-TOPIC-2");

TestUtils.waitForCondition(() -> assignedTopics.equals(expectedSecondAssignment), STREAM_TASKS_NOT_UPDATED);
TestUtils.waitForCondition(() -> assignedTopics.equals(expectedSecondAssignment), STREAM_TASKS_NOT_UPDATED);

streams.close();
CLUSTER.deleteTopicsAndWait("TEST-TOPIC-1", "TEST-TOPIC-2");
streams.close();

@ableegoldman ableegoldman Jun 4, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Should we call close in the finally block? Here and elsewhere

@mjsax mjsax Jun 4, 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.

I don't think that is necessary -- there is an @After method that closed the client for us.

@ableegoldman ableegoldman Jun 4, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Then why close it here as well?

@mjsax mjsax Jun 4, 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.

I think it's better to first close it before we delete the topics.

@ableegoldman ableegoldman Jun 4, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Well, won't we end up deleting the topics before closing it if we never reach the first streams.close ? Or does it not really matter in that case since something has already gone wrong (just curious, I'm fine with it as-is btw)

@mjsax mjsax Jun 4, 2020 •
edited
Loading

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, that is my reasoning.

@guozhangwang guozhangwang Jun 5, 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.

Why we need to call streams.close() inside the function given they are always called in tearDown? Ditto below.

@mjsax mjsax Jun 5, 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.

As mentioned above: we should close the client before we delete the input topics. -- Seems cleaner.

} finally {
CLUSTER.deleteTopicsAndWait("TEST-TOPIC-1", "TEST-TOPIC-2");
}
}

@Test
public void testRegexRecordsAreProcessedAfterReassignment() throws Exception {
final String topic1 = "TEST-TOPIC-1";
CLUSTER.createTopic(topic1);

final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-\\d"));
pattern1Stream.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
streams = new KafkaStreams(builder.build(), streamsConfiguration);
final String topic2 = "TEST-TOPIC-2";
streams.start();

CLUSTER.createTopic(topic2);
try {
CLUSTER.createTopic(topic1);

final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-\\d"));
pattern1Stream.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
streams = new KafkaStreams(builder.build(), streamsConfiguration);
streams.start();

final KeyValue<String, String> record1 = new KeyValue<>("1", "1");
final KeyValue<String, String> record2 = new KeyValue<>("2", "2");
IntegrationTestUtils.produceKeyValuesSynchronously(
CLUSTER.createTopic(topic2);

final KeyValue<String, String> record1 = new KeyValue<>("1", "1");
final KeyValue<String, String> record2 = new KeyValue<>("2", "2");
IntegrationTestUtils.produceKeyValuesSynchronously(
topic1,
Collections.singletonList(record1),
TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
CLUSTER.time
);
IntegrationTestUtils.produceKeyValuesSynchronously(
);
IntegrationTestUtils.produceKeyValuesSynchronously(
topic2,
Collections.singletonList(record2),
TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
CLUSTER.time
);
IntegrationTestUtils.waitUntilFinalKeyValueRecordsReceived(
);
IntegrationTestUtils.waitUntilFinalKeyValueRecordsReceived(
TestUtils.consumerConfig(CLUSTER.bootstrapServers(), StringDeserializer.class, StringDeserializer.class),
outputTopic,
Arrays.asList(record1, record2)
);
);

streams.close();
CLUSTER.deleteTopicsAndWait(topic1, topic2);
streams.close();
} finally {
CLUSTER.deleteTopicsAndWait(topic1, topic2);
}
}

private String createTopic(final int suffix) throws InterruptedException {
Expand All @@ -229,44 +238,44 @@ private String createTopic(final int suffix) throws InterruptedException {

@Test
public void testRegexMatchesTopicsAWhenDeleted() throws Exception {

final Serde<String> stringSerde = Serdes.String();
final List<String> expectedFirstAssignment = Arrays.asList("TEST-TOPIC-A", "TEST-TOPIC-B");
final List<String> expectedSecondAssignment = Collections.singletonList("TEST-TOPIC-B");
final List<String> assignedTopics = new CopyOnWriteArrayList<>();

CLUSTER.createTopics("TEST-TOPIC-A", "TEST-TOPIC-B");
try {
CLUSTER.createTopics("TEST-TOPIC-A", "TEST-TOPIC-B");

final StreamsBuilder builder = new StreamsBuilder();
final StreamsBuilder builder = new StreamsBuilder();

final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-[A-Z]"));
final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-[A-Z]"));

pattern1Stream.to(outputTopic, Produced.with(stringSerde, stringSerde));

final List<String> assignedTopics = new CopyOnWriteArrayList<>();
streams = new KafkaStreams(builder.build(), streamsConfiguration, new DefaultKafkaClientSupplier() {
@Override
public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
return new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
@Override
public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {
super.subscribe(topics, new TheConsumerRebalanceListener(assignedTopics, listener));
}
};
}
});
pattern1Stream.to(outputTopic, Produced.with(stringSerde, stringSerde));

streams = new KafkaStreams(builder.build(), streamsConfiguration, new DefaultKafkaClientSupplier() {
@Override
public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
return new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
@Override
public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {
super.subscribe(topics, new TheConsumerRebalanceListener(assignedTopics, listener));
}
};
}
});

streams.start();
TestUtils.waitForCondition(() -> assignedTopics.equals(expectedFirstAssignment), STREAM_TASKS_NOT_UPDATED);

CLUSTER.deleteTopic("TEST-TOPIC-A");
streams.start();
TestUtils.waitForCondition(() -> assignedTopics.equals(expectedFirstAssignment), STREAM_TASKS_NOT_UPDATED);
} finally {
CLUSTER.deleteTopic("TEST-TOPIC-A");
}

TestUtils.waitForCondition(() -> assignedTopics.equals(expectedSecondAssignment), STREAM_TASKS_NOT_UPDATED);
}

@Test
public void shouldAddStateStoreToRegexDefinedSource() throws InterruptedException {

public void shouldAddStateStoreToRegexDefinedSource() throws Exception {
final ProcessorSupplier<String, String> processorSupplier = new MockProcessorSupplier<>();
final StoreBuilder<KeyValueStore<Object, Object>> storeBuilder = new MockKeyValueStoreBuilder("testStateStore", false);
final long thirtySecondTimeout = 30 * 1000;
Expand All @@ -277,26 +286,19 @@ public void shouldAddStateStoreToRegexDefinedSource() throws InterruptedExceptio
topology.addStateStore(storeBuilder, "my-processor");

streams = new KafkaStreams(topology, streamsConfiguration);
streams.start();

try {
streams.start();

final TestCondition stateStoreNameBoundToSourceTopic = () -> {
final Map<String, List<String>> stateStoreToSourceTopic = topology.getInternalBuilder().stateStoreNameToSourceTopics();
final List<String> topicNamesList = stateStoreToSourceTopic.get("testStateStore");
return topicNamesList != null && !topicNamesList.isEmpty() && topicNamesList.get(0).equals("topic-1");
};

TestUtils.waitForCondition(stateStoreNameBoundToSourceTopic, thirtySecondTimeout, "Did not find topic: [topic-1] connected to state store: [testStateStore]");
final TestCondition stateStoreNameBoundToSourceTopic = () -> {
final Map<String, List<String>> stateStoreToSourceTopic = topology.getInternalBuilder().stateStoreNameToSourceTopics();
final List<String> topicNamesList = stateStoreToSourceTopic.get("testStateStore");
return topicNamesList != null && !topicNamesList.isEmpty() && topicNamesList.get(0).equals("topic-1");
};

} finally {
streams.close();
}
TestUtils.waitForCondition(stateStoreNameBoundToSourceTopic, thirtySecondTimeout, "Did not find topic: [topic-1] connected to state store: [testStateStore]");
}

@Test
public void testShouldReadFromRegexAndNamedTopics() throws Exception {

final String topic1TestMessage = "topic-1 test";
final String topic2TestMessage = "topic-2 test";
final String topicATestMessage = "topic-A test";
Expand Down Expand Up @@ -346,7 +348,6 @@ public void testShouldReadFromRegexAndNamedTopics() throws Exception {

@Test
public void testMultipleConsumersCanReadFromPartitionedTopic() throws Exception {

KafkaStreams partitionedStreamsLeader = null;
KafkaStreams partitionedStreamsFollower = null;
try {
Expand Down Expand Up @@ -401,12 +402,10 @@ public void subscribe(final Pattern topics, final ConsumerRebalanceListener list
partitionedStreamsFollower.close();
}
}

}

@Test
public void testNoMessagesSentExceptionFromOverlappingPatterns() throws Exception {

final String fMessage = "fMessage";
final String fooMessage = "fooMessage";
final Serde<String> stringSerde = Serdes.String();
Expand Down