This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 22fab0d8 [FLINK-40621][Connectors/Kafka] Flush before fencing in
producer reset test (#310)
22fab0d8 is described below
commit 22fab0d8058b044c60c2dca0deb300e9a5a0d027
Author: Sylwester Lachiewicz <[email protected]>
AuthorDate: Sat Oct 24 19:50:16 2026 +0200
[FLINK-40621][Connectors/Kafka] Flush before fencing in producer reset test
(#310)
testResetInnerTransactionIfFinalizingTransactionFailed sent a record
without flushing, then fenced the producer with a second one using the
same transactional id. When the in-flight ProduceRequest reached the
broker after the epoch bump, the broker rejected it with
INVALID_PRODUCER_EPOCH and commitTransaction threw
InvalidProducerEpochException. That is a sibling of
ProducerFencedException under ApplicationRecoverableException, not a
subtype, so the assertion failed.
Flushing before the fencing producer starts keeps the failure on the
EndTxn path, where the broker reports PRODUCER_FENCED. Do not drop the
flush as redundant; it is what makes the outcome deterministic.
Generated-by: Claude Code (Claude Opus 5, Claude Opus 5.5)
---
.../connector/kafka/sink/FlinkKafkaInternalProducerITCase.java | 6 ++++++
1 file changed, 6 insertions(+)
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java
index 77cf3568..9f648208 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java
@@ -159,6 +159,12 @@ class FlinkKafkaInternalProducerITCase {
fenced.initTransactions();
fenced.beginTransaction();
fenced.send(new ProducerRecord<>(topic, "test-value"));
+ // Await the produce before fencing. An in-flight ProduceRequest
that reaches the
+ // broker after the epoch bump below is rejected with
INVALID_PRODUCER_EPOCH, and the
+ // finalizer then throws InvalidProducerEpochException, a sibling
of
+ // ProducerFencedException rather than a subtype. Flushing keeps
the fencing on the
+ // EndTxn path, which is the case this test pins.
+ fenced.flush();
// Start a second producer that fences the first one
try (FlinkKafkaInternalProducer<String, String> producer =
new FlinkKafkaInternalProducer<>(getProperties(),
"dummy")) {
