MINOR: Convert last streams join test to TTD by bbejeck · Pull Request #7777 · apache/kafka

bbejeck · 2019-12-04T15:59:34Z

Conversion of the last join integration test to use TTD

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

bbejeck · 2019-12-04T16:02:14Z

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


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.

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?

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?

bbejeck · 2019-12-04T16:03:41Z

Ping @guozhangwang @cadonna for reviews

cadonna

@bbejeck Thanks for the PR

cadonna · 2019-12-04T16:42:45Z

+            TestUtils.waitForCondition(() -> finalResultReached.get(), "Never received expected final result.");
+            final List<TestRecord<Long, String>> output = outputTopic.readRecordsToList();
+
+            assertEquals(output.get(output.size() - 1), updatedExpectedFinalResult);


prop: Use assertThat() here and in the other overload of this method.

cadonna · 2019-12-04T16:47:31Z

                    firstTimestamp + expectedFinalResult.timestamp());
-            checkResult(OUTPUT_TOPIC, updatedExpectedFinalResult, numRecordsExpected);
+
+            TestUtils.waitForCondition(() -> finalResultReached.get(), "Never received expected final result.");


Q: What are we waiting for here? Shouldn't the result be immediately available?

Good call. This was in the original test and I left this in by oversight, removing this lead to a significant cleanup!

cadonna · 2019-12-04T16:49:10Z

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


prop: Should we use MockTime here?
prop: Could you use a more meaningful name for ts?

The above is also valid for the overload.

Ack. Good call on the MockTime!

cadonna · 2019-12-04T17:01:12Z

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


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.

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.

cadonna · 2019-12-04T17:18:47Z

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


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?

cadonna · 2019-12-04T17:34:00Z

BTW: Very nice PR number and I like PRs that remove more lines than they add. :-))

guozhangwang

Thanks @bbejeck , I do not have any major comments; after @cadonna 's comments are addressed I think we can merge it.

guozhangwang · 2019-12-04T18:30:59Z

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


guozhangwang · 2019-12-04T18:34:09Z

-    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 {


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?

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?

Okay, sounds fine.

bbejeck · 2019-12-05T17:35:35Z

@guozhangwang @cadonna thanks for the reviews!

comments addressed.

Reviewers: Bruno Cadonna <bruno@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

bbejeck · 2019-12-05T22:42:12Z

cherry-picked to 2.4

bbejeck added 2 commits December 3, 2019 20:08

MINOR: Convert TableTableJoinIntegrationTest to use TTD

b52aa8c

MINOR: Remove unused variables

820f75d

bbejeck added streams tests Test fixes (including flaky tests) labels Dec 4, 2019

bbejeck commented Dec 4, 2019

View reviewed changes

cadonna reviewed Dec 4, 2019

View reviewed changes

guozhangwang reviewed Dec 4, 2019

View reviewed changes

bbejeck added 3 commits December 5, 2019 11:47

MINOR: Comments switch to MockTime, improve variable name

d393634

MINOR: Final cleanup from comments.

c3ecebd

MINOR: Fix checkstyle

502526b

guozhangwang approved these changes Dec 5, 2019

View reviewed changes

guozhangwang merged commit 04cd145 into apache:trunk Dec 5, 2019

bbejeck added a commit that referenced this pull request Dec 5, 2019

MINOR: Convert last streams join test to TTD (#7777)

2025eb5

Reviewers: Bruno Cadonna <bruno@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

bbejeck deleted the MINOR_convert_last_streams_join_test_to_TTD branch July 10, 2024 13:58

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

		}


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

Conversation

bbejeck commented Dec 4, 2019

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck commented Dec 4, 2019

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cadonna commented Dec 4, 2019

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck commented Dec 5, 2019

Uh oh!

bbejeck commented Dec 5, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants