root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

KAFKA-19511: Fix flaky test HandlingSourceTopicDeletionIntegrationTest.shouldThrowErrorAfterSourceTopicDeleted (#20187)
CI / build (push) Waiting to run Details 

Temporarily fix it by disable the new protocol, will take a deeper look
at it in the consumer protocol.

Reviewers: Matthias J. Sax <matthias@confluent.io>
This commit is contained in:
Jinhe Zhang 2025-07-18 15:51:45 -04:00 committed by GitHub
parent 9b542b6ea2
commit 38e3359446
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 34 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HandlingSourceTopicDeletionIntegrationTest.java
View File

@ -17,7 +17,6 @@
package org.apache.kafka.streams.integration; package org.apache.kafka.streams.integration;
import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.GroupProtocol;
import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KafkaStreams.State; import org.apache.kafka.streams.KafkaStreams.State;
import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsBuilder;
@ -34,12 +33,10 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo; import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException; import java.io.IOException;
import java.util.Locale;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
@ -78,9 +75,8 @@ public class HandlingSourceTopicDeletionIntegrationTest {
CLUSTER.deleteTopics(INPUT_TOPIC, OUTPUT_TOPIC); CLUSTER.deleteTopics(INPUT_TOPIC, OUTPUT_TOPIC);
} }
@ParameterizedTest @Test
@ValueSource(booleans = {false, true}) public void shouldThrowErrorAfterSourceTopicDeleted(final TestInfo testName) throws InterruptedException {
public void shouldThrowErrorAfterSourceTopicDeleted(final boolean useNewProtocol, final TestInfo testName) throws InterruptedException {
final StreamsBuilder builder = new StreamsBuilder(); final StreamsBuilder builder = new StreamsBuilder();
builder.stream(INPUT_TOPIC, Consumed.with(Serdes.Integer(), Serdes.String())) builder.stream(INPUT_TOPIC, Consumed.with(Serdes.Integer(), Serdes.String()))
.to(OUTPUT_TOPIC, Produced.with(Serdes.Integer(), Serdes.String())); .to(OUTPUT_TOPIC, Produced.with(Serdes.Integer(), Serdes.String()));
@ -91,46 +87,46 @@ public class HandlingSourceTopicDeletionIntegrationTest {
final Properties streamsConfiguration = new Properties(); final Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, appId); streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class);
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, NUM_THREADS); streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, NUM_THREADS);
streamsConfiguration.put(StreamsConfig.METADATA_MAX_AGE_CONFIG, 2000); streamsConfiguration.put(StreamsConfig.METADATA_MAX_AGE_CONFIG, 2000);
if (useNewProtocol) {
streamsConfiguration.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
}
final Topology topology = builder.build(); final Topology topology = builder.build();
final KafkaStreams kafkaStreams1 = new KafkaStreams(topology, streamsConfiguration);
final AtomicBoolean calledUncaughtExceptionHandler1 = new AtomicBoolean(false); final AtomicBoolean calledUncaughtExceptionHandler1 = new AtomicBoolean(false);
kafkaStreams1.setUncaughtExceptionHandler(exception -> {
calledUncaughtExceptionHandler1.set(true);
return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
});
kafkaStreams1.start();
final KafkaStreams kafkaStreams2 = new KafkaStreams(topology, streamsConfiguration);
final AtomicBoolean calledUncaughtExceptionHandler2 = new AtomicBoolean(false); final AtomicBoolean calledUncaughtExceptionHandler2 = new AtomicBoolean(false);
kafkaStreams2.setUncaughtExceptionHandler(exception -> {
calledUncaughtExceptionHandler2.set(true);
return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
});
kafkaStreams2.start();
TestUtils.waitForCondition( try (final KafkaStreams kafkaStreams1 = new KafkaStreams(topology, streamsConfiguration);
() -> kafkaStreams1.state() == State.RUNNING && kafkaStreams2.state() == State.RUNNING, final KafkaStreams kafkaStreams2 = new KafkaStreams(topology, streamsConfiguration)) {
TIMEOUT,
() -> "Kafka Streams clients did not reach state RUNNING" kafkaStreams1.setUncaughtExceptionHandler(exception -> {
); calledUncaughtExceptionHandler1.set(true);
return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
});
kafkaStreams1.start();
CLUSTER.deleteTopic(INPUT_TOPIC); kafkaStreams2.setUncaughtExceptionHandler(exception -> {
calledUncaughtExceptionHandler2.set(true);
return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
});
kafkaStreams2.start();
TestUtils.waitForCondition( TestUtils.waitForCondition(
() -> kafkaStreams1.state() == State.ERROR && kafkaStreams2.state() == State.ERROR, () -> kafkaStreams1.state() == State.RUNNING && kafkaStreams2.state() == State.RUNNING,
TIMEOUT, TIMEOUT,
() -> "Kafka Streams clients did not reach state ERROR" () -> "Kafka Streams clients did not reach state RUNNING"
); );
assertThat(calledUncaughtExceptionHandler1.get(), is(true)); CLUSTER.deleteTopic(INPUT_TOPIC);
assertThat(calledUncaughtExceptionHandler2.get(), is(true));
TestUtils.waitForCondition(
() -> kafkaStreams1.state() == State.ERROR && kafkaStreams2.state() == State.ERROR,
TIMEOUT,
() -> "Kafka Streams clients did not reach state ERROR"
);
assertThat(calledUncaughtExceptionHandler1.get(), is(true));
assertThat(calledUncaughtExceptionHandler2.get(), is(true));
}
} }
} }