This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch v3.4
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/v3.4 by this push:
new 70665ffa [FLINK-39699][tests] Stabilize
KafkaWriterFaultToleranceITCase exception-on-unavailable tests
70665ffa is described below
commit 70665ffab81636ec13c6a5f8dfdcc4503ce4fccf
Author: MartijnVisser <[email protected]>
AuthorDate: Mon Sep 7 20:16:28 2026 +0200
[FLINK-39699][tests] Stabilize KafkaWriterFaultToleranceITCase
exception-on-unavailable tests
Manual port of 887d5941 (warm-up write and flush before stopping the
broker) and the assertion loosening from FLINK-37611 (d74a7bd8, 2b199b0c)
to the v3.4 file layout. The producer may surface either a
NetworkException or a TimeoutException after the broker stops.
Generated-by: Claude Code 2.1.263 (Claude Fable 5.1)
---
.../kafka/sink/KafkaWriterFaultToleranceITCase.java | 15 ++++++++++++---
1 file changed, 12 insertions(+), 3 deletions(-)
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterFaultToleranceITCase.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterFaultToleranceITCase.java
index 1cf1b5c0..b1cf23ad 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterFaultToleranceITCase.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterFaultToleranceITCase.java
@@ -68,13 +68,16 @@ public class KafkaWriterFaultToleranceITCase extends
KafkaWriterTestBase {
properties, DeliveryGuarantee.AT_LEAST_ONCE,
metricGroup)) {
writer.write(1, SINK_WRITER_CONTEXT);
+ writer.getCurrentProducer().flush();
KAFKA_CONTAINER.stop();
try {
+ writer.write(1, SINK_WRITER_CONTEXT);
writer.getCurrentProducer().flush();
assertThatCode(() -> writer.write(1, SINK_WRITER_CONTEXT))
- .hasRootCauseExactlyInstanceOf(NetworkException.class);
+ .rootCause()
+ .isInstanceOfAny(NetworkException.class,
TimeoutException.class);
} finally {
KAFKA_CONTAINER.start();
}
@@ -91,11 +94,14 @@ public class KafkaWriterFaultToleranceITCase extends
KafkaWriterTestBase {
createWriterWithConfiguration(
properties, DeliveryGuarantee.AT_LEAST_ONCE,
metricGroup)) {
writer.write(1, SINK_WRITER_CONTEXT);
+ writer.flush(false);
KAFKA_CONTAINER.stop();
try {
+ writer.write(1, SINK_WRITER_CONTEXT);
assertThatCode(() -> writer.flush(false))
- .hasRootCauseExactlyInstanceOf(NetworkException.class);
+ .rootCause()
+ .isInstanceOfAny(NetworkException.class,
TimeoutException.class);
} finally {
KAFKA_CONTAINER.start();
}
@@ -113,14 +119,17 @@ public class KafkaWriterFaultToleranceITCase extends
KafkaWriterTestBase {
properties, DeliveryGuarantee.AT_LEAST_ONCE,
metricGroup);
writer.write(1, SINK_WRITER_CONTEXT);
+ writer.getCurrentProducer().flush();
KAFKA_CONTAINER.stop();
try {
+ writer.write(1, SINK_WRITER_CONTEXT);
writer.getCurrentProducer().flush();
// closing producer resource throws exception first
assertThatCode(() -> writer.close())
- .hasRootCauseExactlyInstanceOf(NetworkException.class);
+ .rootCause()
+ .isInstanceOfAny(NetworkException.class,
TimeoutException.class);
} catch (Exception e) {
writer.close();
throw e;
