KAFKA-5362: Add EOS system tests for Streams API #3201

mjsax · 2017-06-01T23:33:20Z

No description provided.

mjsax · 2017-06-01T23:33:37Z

Call for review @guozhangwang @bbejeck @dguy

asfbot · 2017-06-02T01:15:32Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4771/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-06-02T01:31:58Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4755/
Test PASSed (JDK 8 and Scala 2.12).

dguy

I'm wondering if we can simplify this test to only do a single operation, i.e., sum? Do we really need to have four different operations to prove that it works? The test seems complicated by the fact that it is trying to verify the various different results and, to me at least, it is adding a bunch of noise that i'm struggling to see through

dguy · 2017-06-02T09:20:58Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

+
+    private static Map<String, Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>> getOutputRecords(final KafkaConsumer<byte[], byte[]> consumer,
+                                                                                                           final Map<TopicPartition, Long> committedOffsets,
+                                                                                                           final List<TopicPartition> partitions) {


partitions is not used

dguy · 2017-06-02T09:22:17Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

+                                                                                                           final List<TopicPartition> partitions) {
+        final Map<String, Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>> recordPerTopicPerPartition = new HashMap<>();
+
+        long maxWaitTime = System.currentTimeMillis() + 30000;


extract 30000 to a constant maybe?

mjsax · 2017-06-02T17:11:00Z

I designed the test similar to SmokeTest that does even more stuff... I still think, it makes sense to test simple "copy" and an "aggregation". I also want to make the test more complex and add a repartitioning step for one of the aggregates -- but it did not work yet, so I just removed the repartitioning...

Maybe we should split this into individual test though. WDYT?

dguy · 2017-06-02T17:17:18Z

I guess it becomes a trade off between system test runtime and code complexity. My preference would be to have simple tests that are easy to follow. If time becomes a problem then merge them.

guozhangwang · 2017-06-02T22:44:53Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestClient.java

+        props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 2);
+        props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
+        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
+        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);


Why do we want to disable caching? I think it is valuable to have at least one test case with caching turned on and see if EOS is still guaranteed?

With caching, we cannot know the result (ie, not each individual record that get's written to the output), as we don't know when we will flush. For this test, I want to check all intermediate records, too, so I need to disable caching.

But we can add a second test with caching and just make sure we don't over count for example and ignore all intermediate results and just check the latest result per key.

guozhangwang · 2017-06-02T22:47:11Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

+        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
+        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
+        producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);


Do you need to set the txnId also?

We write non-transactional data. Actually, we might not even need idempotent producer as we assume that there is no error writing data for this test cases.

guozhangwang · 2017-06-02T22:54:26Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

+
+        for (final Map.Entry<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> partitionRecords : receivedRecords.entrySet()) {
+            final Long committed = committedOffsets.get(partitionRecords.getKey());
+            if (partitionRecords.getValue().size() < (committed == null ? 0 : committed)) {


Is this okay? If committed == null it means no offsets get committed for this partition in getCommittedOffsets, and we should check that we do not get any data in partitionRecords but currently this check will always pass even if partitionRecords.getValue().size() > 0.

guozhangwang · 2017-06-02T22:55:29Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

+
+            verifyAllTransactionFinished(consumer, kafka);
+
+            // no not modify: required test output


no -> do?

guozhangwang · 2017-06-02T22:56:13Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

+
+        for (final Map.Entry<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> partitionRecords : topicRecords.entrySet()) {
+            final TopicPartition tp = partitionRecords.getKey();
+            truncatedTopicRecords.put(tp, partitionRecords.getValue().subList(0, committedOffsets.get(tp).intValue()));


Similarly, committedOffsets.get(tp).intValue() could throw NPE?

Some further improvements

mjsax · 2017-06-07T22:41:46Z

Updated. 3 consecutive local runs passed.

mjsax · 2017-06-07T22:44:35Z

Triggered branch builder at: https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/306/

bbejeck

LGTM

asfbot · 2017-06-07T23:38:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5031/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-06-08T00:06:17Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5047/
Test PASSed (JDK 7 and Scala 2.11).

mjsax · 2017-06-08T00:26:11Z

Branch builder failed... Investigating.

asfbot · 2017-06-08T03:06:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5054/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-06-08T03:11:29Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5038/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-06-08T19:52:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5059/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-06-08T20:00:55Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5075/
Test PASSed (JDK 7 and Scala 2.11).

mjsax · 2017-06-08T20:17:56Z

Branch builder passed: https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/311/

Author: Matthias J. Sax <matthias@confluent.io> Reviewers: Damian Guy <damian.guy@gmail.com>, Bill Bejeck <bill@confluent.io>, Guozhang Wang <wangguoz@gmail.com> Closes #3201 from mjsax/kafka-5362-add-eos-system-tests-for-streams-api (cherry picked from commit ba07d82) Signed-off-by: Guozhang Wang <wangguoz@gmail.com>

guozhangwang · 2017-06-08T21:09:28Z

Merged to trunk and 0.11.0

dguy reviewed Jun 2, 2017

View reviewed changes

guozhangwang reviewed Jun 2, 2017

View reviewed changes

mjsax added 3 commits June 7, 2017 12:48

KAFKA-5362: Add EOS system tests for Streams API

3396941

cleanup

f9fa5d0

Github comments

925da6c

Some further improvements

mjsax force-pushed the kafka-5362-add-eos-system-tests-for-streams-api branch from 321c6a2 to 925da6c Compare June 7, 2017 22:41

bbejeck approved these changes Jun 7, 2017

View reviewed changes

fixed test

842c542

second fix for failing test

088cfe0

asfgit closed this in ba07d82 Jun 8, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-5362: Add EOS system tests for Streams API #3201

KAFKA-5362: Add EOS system tests for Streams API #3201

mjsax commented Jun 1, 2017

mjsax commented Jun 1, 2017

asfbot commented Jun 2, 2017

asfbot commented Jun 2, 2017

dguy left a comment

dguy Jun 2, 2017

dguy Jun 2, 2017

mjsax commented Jun 2, 2017

dguy commented Jun 2, 2017

guozhangwang Jun 2, 2017

mjsax Jun 2, 2017

guozhangwang Jun 2, 2017

mjsax Jun 2, 2017

guozhangwang Jun 2, 2017

guozhangwang Jun 2, 2017

guozhangwang Jun 2, 2017

mjsax commented Jun 7, 2017

mjsax commented Jun 7, 2017

bbejeck left a comment

asfbot commented Jun 7, 2017

asfbot commented Jun 8, 2017

mjsax commented Jun 8, 2017

asfbot commented Jun 8, 2017

asfbot commented Jun 8, 2017

asfbot commented Jun 8, 2017

asfbot commented Jun 8, 2017

mjsax commented Jun 8, 2017

guozhangwang commented Jun 8, 2017


		verifyAllTransactionFinished(consumer, kafka);

		// no not modify: required test output

KAFKA-5362: Add EOS system tests for Streams API #3201

KAFKA-5362: Add EOS system tests for Streams API #3201

Conversation

mjsax commented Jun 1, 2017

mjsax commented Jun 1, 2017

asfbot commented Jun 2, 2017

asfbot commented Jun 2, 2017

dguy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Jun 2, 2017

dguy commented Jun 2, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Jun 7, 2017

mjsax commented Jun 7, 2017

bbejeck left a comment

Choose a reason for hiding this comment

asfbot commented Jun 7, 2017

asfbot commented Jun 8, 2017

mjsax commented Jun 8, 2017

asfbot commented Jun 8, 2017

asfbot commented Jun 8, 2017

asfbot commented Jun 8, 2017

asfbot commented Jun 8, 2017

mjsax commented Jun 8, 2017

guozhangwang commented Jun 8, 2017