Skip to content

MINOR: Convert last streams join test to TTD #7777

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 5 commits into from
Dec 5, 2019
Merged

MINOR: Convert last streams join test to TTD #7777

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b52aa8c
MINOR: Convert TableTableJoinIntegrationTest to use TTD
bbejeck Dec 4, 2019
820f75d
MINOR: Remove unused variables
bbejeck Dec 4, 2019
d393634
MINOR: Comments switch to MockTime, improve variable name
bbejeck Dec 5, 2019
c3ecebd
MINOR: Final cleanup from comments.
bbejeck Dec 5, 2019
502526b
MINOR: Fix checkstyle
bbejeck Dec 5, 2019
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
152 changes: 30 additions & 122 deletions streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,27 +17,21 @@
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.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
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.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.test.TestRecord;
Expand All @@ -52,7 +46,6 @@
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
Expand All @@ -64,10 +57,9 @@
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.apache.kafka.test.StreamsTestUtils.startKafkaStreamsAndWaitForRunningState;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertEquals;
import static org.hamcrest.core.IsEqual.equalTo;

/**
* Tests all available joins of Kafka Streams DSL.
Expand All @@ -92,19 +84,14 @@ public static Collection<Object[]> data() {

static String appID;

private final MockTime time = new MockTime();
private static final Long COMMIT_INTERVAL = 100L;
static final Properties STREAMS_CONFIG = new Properties();
static final String INPUT_TOPIC_RIGHT = "inputTopicRight";
static final String INPUT_TOPIC_LEFT = "inputTopicLeft";
static final String OUTPUT_TOPIC = "outputTopic";
static final long ANY_UNIQUE_KEY = 0L;

private final static Properties PRODUCER_CONFIG = new Properties();
private final static Properties RESULT_CONSUMER_CONFIG = new Properties();

private KafkaProducer<Long, String> producer;
private KafkaStreams streams;

StreamsBuilder builder;
int numRecordsExpected = 0;
AtomicBoolean finalResultReached = new AtomicBoolean(false);
Expand All @@ -131,25 +118,12 @@ public static Collection<Object[]> data() {

final boolean cacheEnabled;

private static final long TIMEOUT = 30000;

AbstractJoinIntegrationTest(final boolean cacheEnabled) {
this.cacheEnabled = cacheEnabled;
}

@BeforeClass
public static void setupConfigsAndUtils() {
PRODUCER_CONFIG.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
PRODUCER_CONFIG.put(ProducerConfig.ACKS_CONFIG, "all");
PRODUCER_CONFIG.put(ProducerConfig.RETRIES_CONFIG, 0);
PRODUCER_CONFIG.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
PRODUCER_CONFIG.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

RESULT_CONSUMER_CONFIG.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
RESULT_CONSUMER_CONFIG.put(ConsumerConfig.GROUP_ID_CONFIG, appID + "-result-consumer");
RESULT_CONSUMER_CONFIG.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
RESULT_CONSUMER_CONFIG.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
RESULT_CONSUMER_CONFIG.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

STREAMS_CONFIG.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
STREAMS_CONFIG.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
Expand All @@ -166,35 +140,12 @@ void prepareEnvironment() throws InterruptedException {
}

STREAMS_CONFIG.put(StreamsConfig.STATE_DIR_CONFIG, testFolder.getRoot().getPath());

producer = new KafkaProducer<>(PRODUCER_CONFIG);
}

@After
public void cleanup() throws InterruptedException {
producer.close(Duration.ofMillis(0));
CLUSTER.deleteAllTopicsAndWait(120000);

@bbejeck bbejeck Dec 4, 2019

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.

This is still needed. We have a couple of tests (non-processing) that start a Kafka Streams application and need the topics to exist, hence we delete them here.

@cadonna cadonna Dec 4, 2019

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.

Q: Does this mean, we still start an embedded Kafka before each test? Is it possible to start an embedded Kafka only in the tests that need it?

@guozhangwang guozhangwang Dec 4, 2019

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.

+1

@bbejeck bbejeck Dec 5, 2019

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 attempted to do that initially, but the logic for using the embedded broker is scattered throughout the base class. There are only two single test methods from all the sub-classes requiring the embedded broker, and these tests don't perform joins. My preference is to merge this PR as is and do a follow-up PR https://issues.apache.org/jira/browse/KAFKA-9273 to refactor the single test methods requiring an embedded broker into a separate class.

WDYT?

}

private void checkResult(final String outputTopic, final List<KeyValueTimestamp<Long, String>> expectedResult) throws InterruptedException {
IntegrationTestUtils.verifyKeyValueTimestamps(RESULT_CONSUMER_CONFIG, outputTopic, expectedResult);
}

private void checkResult(final String outputTopic, final KeyValueTimestamp<Long, String> expectedFinalResult, final int expectedTotalNumRecords) throws InterruptedException {
final List<KeyValueTimestamp<Long, String>> result =
IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(RESULT_CONSUMER_CONFIG, outputTopic, expectedTotalNumRecords, 30 * 1000L);
assertThat(result.get(result.size() - 1), is(expectedFinalResult));
}

/*
* Runs the actual test. Checks the result after each input record to ensure fixed processing order.
* If an input tuple does not trigger any result, "expectedResult" should contain a "null" entry
*/
void runTest(final List<List<KeyValueTimestamp<Long, String>>> expectedResult) throws Exception {
runTest(expectedResult, null);
}


void runTestWithDriver(final List<List<TestRecord<Long, String>>> expectedResult) {

@cadonna cadonna Dec 4, 2019

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.

prop: Would it make sense to also have an overload with just a flat list, i.e.,

void runTestWithDriver(final List<TestRecord<Long, String>> expectedResult)

Maybe it would simplify the code of some of the tests. Hopefully, you can share some of the code in the overloads.

@bbejeck bbejeck Dec 5, 2019

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.

Maybe, but as the List<List<..>> represents the expected results from a join without caching enabled (multiple results) I think that will require a refactoring of the entire test(s). This logic was pre-existing from the original test. Giving that this PR is to convert from using the embedded broker to the TTD to perform the joins, I'd prefer to do the refactoring in a separate PR.

runTestWithDriver(expectedResult, null);
}
Expand All @@ -211,11 +162,11 @@ void runTestWithDriver(final List<List<TestRecord<Long, String>>> expectedResult

TestRecord<Long, String> expectedFinalResult = null;

final long firstTimestamp = System.currentTimeMillis();
long ts = firstTimestamp;
final long firstTimestamp = time.milliseconds();
long eventTimestamp = firstTimestamp;
final Iterator<List<TestRecord<Long, String>>> resultIterator = expectedResult.iterator();
for (final Input<String> singleInputRecord : input) {
testInputTopicMap.get(singleInputRecord.topic).pipeInput(singleInputRecord.record.key, singleInputRecord.record.value, ++ts);
testInputTopicMap.get(singleInputRecord.topic).pipeInput(singleInputRecord.record.key, singleInputRecord.record.value, ++eventTimestamp);

final List<TestRecord<Long, String>> expected = resultIterator.next();
if (expected != null) {
Expand All @@ -225,96 +176,53 @@ void runTestWithDriver(final List<List<TestRecord<Long, String>>> expectedResult
}

final List<TestRecord<Long, String>> output = outputTopic.readRecordsToList();
assertEquals(output, updatedExpected);
expectedFinalResult = updatedExpected.get(expected.size() - 1);
}
}
}
}


/*
* Runs the actual test. Checks the result after each input record to ensure fixed processing order.
* If an input tuple does not trigger any result, "expectedResult" should contain a "null" entry
*/
void runTest(final List<List<KeyValueTimestamp<Long, String>>> expectedResult, final String storeName) throws Exception {
assert expectedResult.size() == input.size();

IntegrationTestUtils.purgeLocalStreamsState(STREAMS_CONFIG);
streams = new KafkaStreams(builder.build(), STREAMS_CONFIG);

KeyValueTimestamp<Long, String> expectedFinalResult = null;

try {
startKafkaStreamsAndWaitForRunningState(streams, TIMEOUT);

final long firstTimestamp = System.currentTimeMillis();
long ts = firstTimestamp;

final Iterator<List<KeyValueTimestamp<Long, String>>> resultIterator = expectedResult.iterator();
for (final Input<String> singleInput : input) {
producer.send(new ProducerRecord<>(singleInput.topic, null, ++ts, singleInput.record.key, singleInput.record.value)).get();

final List<KeyValueTimestamp<Long, String>> expected = resultIterator.next();

if (expected != null) {
final List<KeyValueTimestamp<Long, String>> updatedExpected = new LinkedList<>();
for (final KeyValueTimestamp<Long, String> record : expected) {
updatedExpected.add(new KeyValueTimestamp<>(record.key(), record.value(), firstTimestamp + record.timestamp()));
}

checkResult(OUTPUT_TOPIC, updatedExpected);
assertThat(output, equalTo(updatedExpected));
expectedFinalResult = updatedExpected.get(expected.size() - 1);
}
}

if (storeName != null) {
checkQueryableStore(storeName, expectedFinalResult);
checkQueryableStore(storeName, expectedFinalResult, driver);
}
} finally {
streams.close();
}
}

/*
* Runs the actual test. Checks the final result only after expected number of records have been consumed.
*/
void runTest(final KeyValueTimestamp<Long, String> expectedFinalResult, final String storeName) throws Exception {
IntegrationTestUtils.purgeLocalStreamsState(STREAMS_CONFIG);
streams = new KafkaStreams(builder.build(), STREAMS_CONFIG);
void runTestWithDriver(final TestRecord<Long, String> expectedFinalResult, final String storeName) throws InterruptedException {

@guozhangwang guozhangwang Dec 4, 2019

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.

This is not introduced in this PR but in the existing runTestWithDriver we never use the passed in storeName it seems, and also the local expectedFinalResult are not used. Is that intentional?

@bbejeck bbejeck Dec 5, 2019

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.

The TableTableJoinIntegrationTest passes in the store name as a parameter and it's used on line 218.
The test uses the expectedFinalResult parameter on lines 205-210. Having said that, this test could use an overall refactoring (cf https://issues.apache.org/jira/browse/KAFKA-9273), but I'd prefer to do so in a follow-up PR.
WDYT?

@guozhangwang guozhangwang Dec 5, 2019

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.

Okay, sounds fine.

try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(STREAMS_CONFIG), STREAMS_CONFIG)) {
final TestInputTopic<Long, String> right = driver.createInputTopic(INPUT_TOPIC_RIGHT, new LongSerializer(), new StringSerializer());
final TestInputTopic<Long, String> left = driver.createInputTopic(INPUT_TOPIC_LEFT, new LongSerializer(), new StringSerializer());
final TestOutputTopic<Long, String> outputTopic = driver.createOutputTopic(OUTPUT_TOPIC, new LongDeserializer(), new StringDeserializer());
final Map<String, TestInputTopic<Long, String>> testInputTopicMap = new HashMap<>();

try {
startKafkaStreamsAndWaitForRunningState(streams, TIMEOUT);
testInputTopicMap.put(INPUT_TOPIC_RIGHT, right);
testInputTopicMap.put(INPUT_TOPIC_LEFT, left);

final long firstTimestamp = System.currentTimeMillis();
long ts = firstTimestamp;
final long firstTimestamp = time.milliseconds();
long eventTimestamp = firstTimestamp;

for (final Input<String> singleInput : input) {
producer.send(new ProducerRecord<>(singleInput.topic, null, ++ts, singleInput.record.key, singleInput.record.value)).get();
for (final Input<String> singleInputRecord : input) {
testInputTopicMap.get(singleInputRecord.topic).pipeInput(singleInputRecord.record.key, singleInputRecord.record.value, ++eventTimestamp);
}

TestUtils.waitForCondition(() -> finalResultReached.get(), "Never received expected final result.");

final KeyValueTimestamp<Long, String> updatedExpectedFinalResult =
new KeyValueTimestamp<>(
final TestRecord<Long, String> updatedExpectedFinalResult =
new TestRecord<Long, String>(
expectedFinalResult.key(),
expectedFinalResult.value(),
null,
firstTimestamp + expectedFinalResult.timestamp());
checkResult(OUTPUT_TOPIC, updatedExpectedFinalResult, numRecordsExpected);

final List<TestRecord<Long, String>> output = outputTopic.readRecordsToList();

assertThat(output.get(output.size() - 1), equalTo(updatedExpectedFinalResult));

if (storeName != null) {
checkQueryableStore(storeName, updatedExpectedFinalResult);
checkQueryableStore(storeName, updatedExpectedFinalResult, driver);
}
} finally {
streams.close();
}
}

/*
* Checks the embedded queryable state store snapshot
*/
private void checkQueryableStore(final String queryableName, final KeyValueTimestamp<Long, String> expectedFinalResult) {
final ReadOnlyKeyValueStore<Long, ValueAndTimestamp<String>> store = streams.store(queryableName, QueryableStoreTypes.timestampedKeyValueStore());
private void checkQueryableStore(final String queryableName, final TestRecord<Long, String> expectedFinalResult, final TopologyTestDriver driver) {
final ReadOnlyKeyValueStore<Long, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore(queryableName);

final KeyValueIterator<Long, ValueAndTimestamp<String>> all = store.all();
final KeyValue<Long, ValueAndTimestamp<String>> onlyEntry = all.next();
Expand Down
Loading