You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chenyulin0719 (via GitHub)" <gi...@apache.org> on 2024/09/13 08:45:15 UTC
[PR] KAFKA-17515: Fix flaky RestoreIntegrationTest.shouldInvokeUserDefinedGlobalStateRestoreListener [kafka]
chenyulin0719 opened a new pull request, #17187:
URL: https://github.com/apache/kafka/pull/17187
It's regarding [KAFKA-17515](https://issues.apache.org/jira/browse/KAFKA-17515).
Found two issues in the flaky tests: (Put the log analysis under Jira comments.)
1. The local state is purged immediately after kafkaStreams.close() timeout (Current timeout = 5 sec). However, the asynchronize close() thread still trigger RocksDBStore.flush() to the local state dir.
2. Racing issue: Task to-be restored in `ks-1` are rebalanced to `ks-2` before entering active restoring state. So no onRestoreSuspend() was triggered.
To solve the issues:
1. Extend the kafkaStreams.close() timeout from 5 sec to 60 sec
2. Ensure all tasks in `ks-1` are active restoring before start second KafkaStreams(`ks-2`)
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17515: Fix flaky RestoreIntegrationTest.shouldInvokeUserDefinedGlobalStateRestoreListener [kafka]
Posted by "chenyulin0719 (via GitHub)" <gi...@apache.org>.
chenyulin0719 commented on code in PR #17187:
URL: https://github.com/apache/kafka/pull/17187#discussion_r1764864632
##########
streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -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));
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);
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17515: Fix flaky RestoreIntegrationTest.shouldInvokeUserDefinedGlobalStateRestoreListener [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #17187:
URL: https://github.com/apache/kafka/pull/17187
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17515: Fix flaky RestoreIntegrationTest.shouldInvokeUserDefinedGlobalStateRestoreListener [kafka]
Posted by "chenyulin0719 (via GitHub)" <gi...@apache.org>.
chenyulin0719 commented on code in PR #17187:
URL: https://github.com/apache/kafka/pull/17187#discussion_r1764864632
##########
streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -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));
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);
Review Comment:
Updated. I tried to rephrase it in my word. Please let me know if it's still not clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17515: Fix flaky RestoreIntegrationTest.shouldInvokeUserDefinedGlobalStateRestoreListener [kafka]
Posted by "chenyulin0719 (via GitHub)" <gi...@apache.org>.
chenyulin0719 commented on code in PR #17187:
URL: https://github.com/apache/kafka/pull/17187#discussion_r1764866480
##########
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java:
##########
@@ -634,6 +634,19 @@ public static void waitForStandbyCompletion(final KafkaStreams streams,
}
}
+ /**
+ * Wait until enough restoring tasks have been started
+ */
+ public static void waitForActiveRestoringTask(final KafkaStreams streams,
+ final int expectedTasks,
+ final long timeoutMilliseconds) throws Exception {
+ TestUtils.waitForCondition(() -> {
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17515: Fix flaky RestoreIntegrationTest.shouldInvokeUserDefinedGlobalStateRestoreListener [kafka]
Posted by "chenyulin0719 (via GitHub)" <gi...@apache.org>.
chenyulin0719 commented on code in PR #17187:
URL: https://github.com/apache/kafka/pull/17187#discussion_r1764864632
##########
streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -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));
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);
Review Comment:
Updated. I tried to rephrase it in my words. Please let me know if it's still not clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17515: Fix flaky RestoreIntegrationTest.shouldInvokeUserDefinedGlobalStateRestoreListener [kafka]
Posted by "chenyulin0719 (via GitHub)" <gi...@apache.org>.
chenyulin0719 commented on code in PR #17187:
URL: https://github.com/apache/kafka/pull/17187#discussion_r1764866283
##########
streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -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));
Review Comment:
It make sense to me. Thanks!
Removed the timeout.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17515: Fix flaky RestoreIntegrationTest.shouldInvokeUserDefinedGlobalStateRestoreListener [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17187:
URL: https://github.com/apache/kafka/pull/17187#discussion_r1760120025
##########
streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -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));
Review Comment:
this test has timeout, so maybe we can remove the timeout from close?
##########
streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -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));
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);
Review 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`
##########
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java:
##########
@@ -634,6 +634,19 @@ public static void waitForStandbyCompletion(final KafkaStreams streams,
}
}
+ /**
+ * Wait until enough restoring tasks have been started
+ */
+ public static void waitForActiveRestoringTask(final KafkaStreams streams,
+ final int expectedTasks,
+ final long timeoutMilliseconds) throws Exception {
+ TestUtils.waitForCondition(() -> {
Review Comment:
Could you please use lambda function?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org