[
https://issues.apache.org/jira/browse/FLINK-40624?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18114060#comment-18114060
]
Bhanu Chander Vallabaneni commented on FLINK-40624:
---------------------------------------------------
[~slachiewicz] are you already working on this one? You filed it, so I do not
want to take it out from under you -- if it is unclaimed I am happy to pick it
up.
The description made the shape clear enough to check against {{main}}:
{{InvalidProducerEpochException}} appears nowhere in flink-connector-kafka, so
the produce-path fencing falls past the {{ProducerFencedException}} branch in
{{KafkaCommitter}} into the terminal {{catch (Exception)}} and is reported
through {{signalFailedWithUnknownReason}} with the "data has been potentially
lost" message, rather than the fenced-specific path. In kafka-clients 4.2.0 the
two are siblings under {{ApplicationRecoverableException}}, so catching one
does not cover the other.
If you would like me to take it, could a committer assign it to me? I have been
working in this repo on FLINK-40586 (PR #309). Otherwise happy to leave it with
you and review.
> KafkaCommitter and ExactlyOnceKafkaWriter do not handle
> InvalidProducerEpochException alongside ProducerFencedException
> -----------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40624
> URL: https://issues.apache.org/jira/browse/FLINK-40624
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Reporter: Sylwester Lachiewicz
> Priority: Minor
>
> A fenced Kafka producer can surface either of two exceptions, depending on
> which request observes the epoch bump:
> * coordinator path (EndTxn, AddPartitions) -> PRODUCER_FENCED ->
> {{ProducerFencedException}}
> * produce path -> INVALID_PRODUCER_EPOCH -> {{InvalidProducerEpochException}}
> In kafka-clients 4.2.0 both extend {{ApplicationRecoverableException}} and
> neither is a subtype of the other. {{InvalidProducerEpochException}} does not
> appear anywhere in flink-connector-kafka, so only the first is handled:
> *
> [KafkaCommitter.java:109|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java#L109]
> catches {{ProducerFencedException}} and routes it to {{logFencedRequest}}
> plus {{signalFailedWithKnownReason}}. {{InvalidProducerEpochException}}
> misses that branch and every later one, falling through to the terminal
> {{catch (Exception e)}} at
> [KafkaCommitter.java:141|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java#L141],
> which logs "Transaction ({}) encountered error and data has been potentially
> lost" and calls {{signalFailedWithUnknownReason}}. A fenced producer is then
> reported as an unknown error and the fenced-specific guidance is not emitted.
> *
> [ExactlyOnceKafkaWriter.java:288|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/ExactlyOnceKafkaWriter.java#L288]
> catches {{ProducerFencedException}} to keep the abort best effort, as its
> comment describes. {{InvalidProducerEpochException}} escapes that abort
> instead.
> This is a handling and reporting gap rather than a correctness bug in the
> commit itself; the terminal branch still triggers failover.
> Evidence is the class hierarchy plus the catch chains, and
> [FLINK-40621|https://issues.apache.org/jira/browse/FLINK-40621], where
> {{commitTransaction}} on a fenced producer threw
> {{InvalidProducerEpochException}} with the produce-path message "attempted to
> produce with an old epoch". I have not reproduced the production path.
> Suggested change: handle both exceptions wherever the intent is "the producer
> is fenced", either by catching both or by catching the shared supertype.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
