KAFKA-14572: Migrate EmbeddedKafkaCluster used by Streams integration tests from EmbeddedZookeeper to KRaft

[OmniaGM]

Committer Checklist (excluded from commit message)

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

[OmniaGM]

Still fixing few test cases please don't review yet

[OmniaGM]

@lucasbru I have 4 test cases failing and most of them seem to be related to AT_LEAST_ONCE feature. I can see that when we try to consume data from output topic we don't get all the records back. During debugging I noticed that all records potentially make it to the output topic but it take long time and I don't want to increase the timeout. Is there is anyway in stream to check if all records made it before pulling the records? Or any suggestions here?

[OmniaGM]

@lucasbru I have 4 test cases failing and most of them seem to be related to AT_LEAST_ONCE feature. I can see that when we try to consume data from output topic we don't get all the records back. During debugging I noticed that all records potentially make it to the output topic but it take long time and I don't want to increase the timeout. Is there is anyway in stream to check if all records made it before pulling the records? Or any suggestions here?

I changed the order for {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE, StreamsConfig.EXACTLY_ONCE_V2} to {StreamsConfig.EXACTLY_ONCE, StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2} and I don't think it's related to AT_LEAST_ONCE per say but whatever test come first will have a problem polling the records correctly.

[lucasbru]

@OmniaGM Looks like something in kraft is just taking longer? Have we not observed this problem with the connect unit tests? I wonder if there are some settings that we could use to speed things up.

[lucasbru]

Could we try non-combined mode with this?

[OmniaGM]

@OmniaGM Looks like something in kraft is just taking longer? Have we not observed this problem with the connect unit tests? I wonder if there are some settings that we could use to speed things up.

It did and I believe the solution was to wait for a warmup producer and consumer to work before any tests

Could we try non-combined mode with this?

Will give it a try.

[OmniaGM]

Could we try non-combined mode with this?

I have tried the non-combined mode but nothing changed. I also tried to make sure the cluster is warmed up by producing to each broker but still didn't make any diff. I am just confused because for test like JoinGracePeriodDurabilityIntegrationTest we are able to publish and consume normally till we restart the stream and then we fail because we are expecting 3 but we are getting 2 records. So am confused why before restarting the stream all tests passes but restarting everything fails only on the first run of the test case

[lucasbru]

EDIT: that's probably not the problem :) but I don't quite understand the problem either

I checked JoinGracePeriodDurabilityIntegrationTest. This one seems to test something specifically about how time progresses. I couldn't find the problem yet, but I wonder if it's somehow related to mocktime... I'll have another look later

[OmniaGM]

EDIT: that's probably not the problem :) but I don't quite understand the problem either

I checked JoinGracePeriodDurabilityIntegrationTest. This one seems to test something specifically about how time progresses. I couldn't find the problem yet, but I wonder if it's somehow related to mocktime... I'll have another look later

I'll try to have another look later this week

[bbejeck]

Hi @OmniaGM - I took a pass and and found the problem.

For the failing tests, the EmbeddedKafkaCluster started with time 0; subsequently, the log cleaner would run and delete the log. If we use the constructor overload of the EmbeddedKafkaCluster that doesn't take a time parameter, it will initialize to the current time. The other fix for tests with low timestamp values (0, 5000, etc.) was to create a static variable with the current time and add it to the timestamp, preserving the test logic. I've pushed a PR to demonstrate the changes for updating this one

I'm still looking at one failure in the RestoreIntegrationTest that I think is unrelated to timestamps. Should be fixed now

[OmniaGM]

Hi @OmniaGM - I took a pass and and found the problem.

For the failing tests, the EmbeddedKafkaCluster started with time 0; subsequently, the log cleaner would run and delete the log. If we use the constructor overload of the EmbeddedKafkaCluster that doesn't take a time parameter, it will initialize to the current time. The other fix for tests with low timestamp values (0, 5000, etc.) was to create a static variable with the current time and add it to the timestamp, preserving the test logic. I've pushed a PR to demonstrate the changes for updating this one

I'm still looking at one failure in the RestoreIntegrationTest that I think is unrelated to timestamps. Should be fixed now

Thanks @bbejeck for spotting this, I updated the pr now

[bbejeck]

@OmniaGM I just looked at the build report from my sample PR, and there were a bunch of failures for the integration tests, where they were all passing locally so we'll see how it goes for you

[bbejeck]

Hey @OmniaGM - great news - I made a few tweeks and the dummy PR I have has 2 green builds for Java 11 and Java 17 . I just noticed the same here, we've done away with the jenkins build so if you update I think we'll have a clean build.

Ping me after updating and we'll get this reviewed and merged!

[OmniaGM]

Hey @OmniaGM - great news - I made a few tweeks and the dummy PR I have has 2 green builds for Java 11 and Java 17 . I just noticed the same here, we've done away with the jenkins build so if you update I think we'll have a clean build.

Ping me after updating and we'll get this reviewed and merged!

Thanks @bbejeck I have pulled your fixes with some changes. I used IntegrationTestUtils.startApplicationAndWaitUntilRunning instead of use

streams.start();
TestUtils.waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING, "KafkaStreams not running in time");

Tests are passing locally will wait for Jenkins

[bbejeck]

@OmniaGM SGTM, thanks for updating

[bbejeck]

Thanks for the contribution @OmniaGM! LGTM

[bbejeck]

Merged #17016 to trunk