Skip to content

KAFKA-8940: decrease session timeout to make test faster and reliable #10871

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guozhangwang merged 1 commit into from
Jun 13, 2021
Merged

KAFKA-8940: decrease session timeout to make test faster and reliable #10871

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions ...ms/src/test/java/org/apache/kafka/streams/integration/SmokeTestDriverIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.integration;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
Expand Down Expand Up @@ -89,6 +90,10 @@ SmokeTestDriver.VerificationResult result() {

}

// In this test, we try to keep creating new stream, and closing the old one, to maintain only 3 streams alive.
// During the new stream added and old stream left, the stream process should still complete without issue.
// We set 2 timeout condition to fail the test before passing the verification:
// (1) 6 min timeout, (2) 30 tries of polling without getting any data
@Test
public void shouldWorkWithRebalance() throws InterruptedException {
Exit.setExitProcedure((statusCode, message) -> {
Expand All @@ -110,6 +115,8 @@ public void shouldWorkWithRebalance() throws InterruptedException {

final Properties props = new Properties();
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
// decrease the session timeout so that we can trigger the rebalance soon after old client left closed
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);

// cycle out Streams instances as long as the test is running.
while (driver.isAlive()) {
Expand Down
7 changes: 4 additions & 3 deletions streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,13 +126,14 @@ public void start(final Properties streamsProperties) {
try {
if (!countDownLatch.await(1, TimeUnit.MINUTES)) {
System.out.println(name + ": SMOKE-TEST-CLIENT-EXCEPTION: Didn't start in one minute");
} else {
System.out.println(name + ": SMOKE-TEST-CLIENT-STARTED");
System.out.println(name + " started at " + Instant.now());
}
} catch (final InterruptedException e) {
System.out.println(name + ": SMOKE-TEST-CLIENT-EXCEPTION: " + e);
e.printStackTrace(System.out);
}
System.out.println(name + ": SMOKE-TEST-CLIENT-STARTED");
System.out.println(name + " started at " + Instant.now());
}

public void closeAsync() {
Expand All @@ -145,7 +146,7 @@ public void close() {
if (wasClosed && !uncaughtException) {
System.out.println(name + ": SMOKE-TEST-CLIENT-CLOSED");
} else if (wasClosed) {
System.out.println(name + ": SMOKE-TEST-CLIENT-EXCEPTION");
System.out.println(name + ": SMOKE-TEST-CLIENT-EXCEPTION: Got an uncaught exception");
} else {
System.out.println(name + ": SMOKE-TEST-CLIENT-EXCEPTION: Didn't close in time.");
}
Expand Down