KAFKA-5362: Add Streams EOS system test with repartitioning topic #3310

mjsax · 2017-06-13T00:46:39Z

No description provided.

mjsax · 2017-06-13T00:47:02Z

Call for review @bbejeck @enothereska @dguy @guozhangwang

mjsax · 2017-06-13T00:48:00Z

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

asfbot · 2017-06-13T01:45:30Z

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

asfbot · 2017-06-13T02:20:27Z

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

mjsax · 2017-06-13T21:06:41Z

Branch builder failed before starting the test... retriggered at https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/324/

mjsax · 2017-06-14T16:03:20Z

Failing system test was due to broker bug that got fixed. Rebased and rerunning system test now: https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/326/

asfbot · 2017-06-14T17:10:08Z

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

asfbot · 2017-06-14T17:20:12Z

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

mjsax · 2017-06-14T22:05:41Z

System test passed: https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/328/
(seems you need to schedule a new build -- rerunning a failed one after a rebase seems to use the old version and does not pick up the change)

guozhangwang · 2017-06-15T16:32:41Z

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

+        if (withRepartitioning) {
+            final KStream<String, Integer> repartitionedData = data.through("repartition");
+
+            repartitionedData.process(SmokeTestUtil.printProcessorSupplier("repartition"));


While checking the SmokeTestUtil code: some of the static variables are not used any more and hence can be removed.

guozhangwang · 2017-06-15T16:48:35Z

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

+
+            final KGroupedStream<String, Integer> groupedData2 = repartitionedData.groupByKey();
+            // max
+            groupedData2


nit: groupedData2 -> groupedDataAfterRepartition

guozhangwang · 2017-06-15T18:45:42Z

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

+    private static boolean receivedAllRecords(final Map<String, Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>> receivedRecords,
+                                              final Map<TopicPartition, Long> committedOffsets,
+                                              final boolean withRepartitioning) {
+        if (receivedRecords == null) {


I think this would never happen now since the passed in recordPerTopicPerPartition is always initialized.

guozhangwang · 2017-06-15T18:48:16Z

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

+        // repartition -> max, cnt
+
+        final Map<String, List<String>> inputToOutputTopics = new HashMap<>();
+        {


This bracket seems not necessary?

I used it to scope the initialization for inputToOutputTopics -- this make the code more readable IMHO as you can easily see where the initialization code "ends".

guozhangwang · 2017-06-15T19:09:35Z

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

+
+                final TopicPartition outputTp = new TopicPartition(outputTopic, partition);
+                final List<ConsumerRecord<byte[], byte[]>> records = received.get(outputTp);
+                if (records == null || records.size() < offset) {


This check records.size() < offset seems a bit sketchy to me. Basically we are assuming that the input topic data's starting offset is always 0, and there is no "holes" in the topic partitions so that the offset indicates the number of records we can read from the input topic.

Maybe a more robust way to do that would be

read the input topics data and optionally repartition based on withRepartitioning, stop when the current record's offset is equal to or larger than the committed offset, and remember the number of records;

read the output topics (again optionally repartition) from the beginning to the end (use seekTo), and check that the number of records are the same as the number of records read from the input.

Then we do not need to truncate, and also in verification we do not need to check list size again since they are already checked here.

mjsax · 2017-06-16T05:06:11Z

Updated. Retriggered branch builder: https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/329/

asfgit · 2017-06-16T05:07:13Z

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

asfgit · 2017-06-16T05:08:02Z

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

mjsax · 2017-06-16T17:30:28Z

Branch builder failed with

Exception in thread "main" java.lang.RuntimeException: Result verification failed for ConsumerRecord(topic = min, partition = 3, offset = 4046, CreateTime = 1497609794842, serialized key size = 2, serialized value size = 4, headers = RecordHeaders(headers = [], isReadOnly = false), key = [B@4f69385e, value = [B@4f3cc0d) expected <47,3854> but was <47,3854>

Note the expected <47,3854> but was <47,3854> -- ??? Will just rerun it again. System tests did pass locally.

New run at: https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/330/

mjsax · 2017-06-16T17:30:36Z

Retest this please.

asfgit · 2017-06-16T17:33:54Z

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

asfgit · 2017-06-16T17:34:02Z

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

asfgit · 2017-06-16T18:55:18Z

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

asfgit · 2017-06-16T19:29:07Z

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

asfgit · 2017-06-16T20:57:14Z

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

asfgit · 2017-06-16T21:20:43Z

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

guozhangwang · 2017-06-16T23:01:53Z

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

+                        throw new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + key + "," + value + "> but was <" + receivedKey + "," + receivedValue + ">");
+                    }
+                }
+            } catch (final NullPointerException e) {


We should check that inputRecords.hasNext is false after the for loop, or any manners to make sure that pairing lists have the same size.

Ditto elsewhere.

guozhangwang · 2017-06-16T23:03:38Z

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

@@ -383,7 +403,84 @@ private static void verifySum(final Map<TopicPartition, List<ConsumerRecord<byte
                    sum += value;


nit: if (sum = value) else (sum += value)

guozhangwang · 2017-06-16T23:04:31Z

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

+                    Integer max = currentMinPerKey.get(key);
+                    if (max == null) {
+                        max = value;
+                    }


nit: if (max = value) else (max = Math.max).

guozhangwang · 2017-06-16T23:06:23Z

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

+                    }
+                }
+            } catch (final NullPointerException e) {
+                System.err.println(inputPerTopicPerPartition);


Could we print both input and output partition lists, and also print as Error found: input is ... output is ....

Ditto elsewhere.

Just thinking about this twice, a NPE should not happen here any more as we reworked the code. IMHO, we can just remove the whole try-catch-block. WDYT?

asfgit · 2017-06-17T02:38:13Z

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

asfgit · 2017-06-17T02:47:05Z

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

ijuma · 2017-06-20T12:18:56Z

What is the status of this?

guozhangwang · 2017-06-20T18:17:27Z

I'm putting this on hold until #3375 is merged, since we also saw a couple local test failures with it and suspected it is due to the same root cause.

asfgit · 2017-06-25T00:12:50Z

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

asfgit · 2017-06-25T00:13:14Z

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

asfgit · 2017-06-25T03:12:32Z

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

asfgit · 2017-06-25T03:23:19Z

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

mjsax · 2017-06-25T03:45:28Z

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

guozhangwang · 2017-06-25T16:35:00Z

LGTM! Merged to trunk and 0.11.0.

Author: Matthias J. Sax <matthias@confluent.io> Reviewers: Guozhang Wang <wangguoz@gmail.com> Closes #3310 from mjsax/kafka-5362-add-eos-system-tests-for-streams-api (cherry picked from commit 2265834) Signed-off-by: Guozhang Wang <wangguoz@gmail.com>

mjsax force-pushed the kafka-5362-add-eos-system-tests-for-streams-api branch from 71afa21 to 435d821 Compare June 14, 2017 16:02

guozhangwang reviewed Jun 15, 2017

View reviewed changes

mjsax force-pushed the kafka-5362-add-eos-system-tests-for-streams-api branch from 435d821 to 5866817 Compare June 16, 2017 05:03

guozhangwang reviewed Jun 16, 2017

View reviewed changes

mjsax added 2 commits June 24, 2017 16:59

KAFKA-5362: Add Streams EOS system test with repartitioning topic

df26f73

Cleanup SmokeTestUtil

d4b6888

mjsax added 4 commits June 24, 2017 16:59

Guozhang's comments

fb21c98

fixed checkstyle

f9b0780

fixing flaky compare

8ae62fd

Guozhang comments

7ee5752

mjsax force-pushed the kafka-5362-add-eos-system-tests-for-streams-api branch from 0c5d9db to 7ee5752 Compare June 25, 2017 00:09

Rebased

fc2762d

asfgit closed this in 2265834 Jun 25, 2017

mjsax deleted the kafka-5362-add-eos-system-tests-for-streams-api branch June 26, 2017 18:23

		@@ -383,7 +403,84 @@ private static void verifySum(final Map<TopicPartition, List<ConsumerRecord<byte
		sum += value;

KAFKA-5362: Add Streams EOS system test with repartitioning topic #3310

KAFKA-5362: Add Streams EOS system test with repartitioning topic #3310

Conversation

mjsax commented Jun 13, 2017

mjsax commented Jun 13, 2017

mjsax commented Jun 13, 2017

asfbot commented Jun 13, 2017

asfbot commented Jun 13, 2017

mjsax commented Jun 13, 2017

mjsax commented Jun 14, 2017

asfbot commented Jun 14, 2017

asfbot commented Jun 14, 2017

mjsax commented Jun 14, 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

mjsax commented Jun 16, 2017

asfgit commented Jun 16, 2017

asfgit commented Jun 16, 2017

mjsax commented Jun 16, 2017 • edited Loading

mjsax commented Jun 16, 2017

asfgit commented Jun 16, 2017

asfgit commented Jun 16, 2017

asfgit commented Jun 16, 2017

asfgit commented Jun 16, 2017

asfgit commented Jun 16, 2017

asfgit commented Jun 16, 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

Choose a reason for hiding this comment

asfgit commented Jun 17, 2017

asfgit commented Jun 17, 2017

ijuma commented Jun 20, 2017

guozhangwang commented Jun 20, 2017

asfgit commented Jun 25, 2017

asfgit commented Jun 25, 2017

asfgit commented Jun 25, 2017

asfgit commented Jun 25, 2017

mjsax commented Jun 25, 2017

guozhangwang commented Jun 25, 2017

mjsax commented Jun 16, 2017 •

edited

Loading