KAFKA-17515: Fix flaky RestoreIntegrationTest.shouldInvokeUserDefinedGlobalStateRestoreListener #17187
Conversation
…GlobalStateRestoreListener
There was a problem hiding this comment.
@chenyulin0719 thanks for digging in this flaky again. a couple of comments are left. PTAL
| @@ -579,12 +580,15 @@ public void shouldInvokeUserDefinedGlobalStateRestoreListener() throws Exception | |||
| validateReceivedMessages(sampleData, outputTopic); | |||
|
|
|||
| // Close kafkaStreams1 (with cleanup) and start it again to force the restoration of the state. | |||
| kafkaStreams.close(Duration.ofMillis(5000L)); | |||
| kafkaStreams.close(Duration.ofMillis(IntegrationTestUtils.DEFAULT_TIMEOUT)); | |||
There was a problem hiding this comment.
this test has timeout, so maybe we can remove the timeout from close?
There was a problem hiding this comment.
It make sense to me. Thanks!
Removed the timeout.
| public static void waitForActiveRestoringTask(final KafkaStreams streams, | ||
| final int expectedTasks, | ||
| final long timeoutMilliseconds) throws Exception { | ||
| TestUtils.waitForCondition(() -> { |
There was a problem hiding this comment.
Could you please use lambda function?
| IntegrationTestUtils.purgeLocalStreamsState(streamsConfigurations); | ||
|
|
||
| final TestStateRestoreListener kafkaStreams1StateRestoreListener = new TestStateRestoreListener("ks1", RESTORATION_DELAY); | ||
| kafkaStreams = startKafkaStreams(builder, kafkaStreams1StateRestoreListener, kafkaStreams1Configuration); | ||
|
|
||
| // Must make ensure all the restoring tasks are in active state before starting the new instance. | ||
| waitForActiveRestoringTask(kafkaStreams, 5, IntegrationTestUtils.DEFAULT_TIMEOUT); |
There was a problem hiding this comment.
Could you include "suspend" in the comments? for example: all restoring task must be in active state. otherwise, the first kafka streams won't encounter "suspend" after starting another instance
There was a problem hiding this comment.
Updated. I tried to rephrase it in my words. Please let me know if it's still not clear.
It's regarding KAFKA-17515.
Found two issues in the flaky tests: (Put the log analysis under Jira comments.)
ks-1are rebalanced toks-2before entering active restoring state. So no onRestoreSuspend() was triggered.To solve the issues:
ks-1are active restoring before start second KafkaStreams(ks-2)Committer Checklist (excluded from commit message)