spuru9 commented on code in PR #314:
URL:
https://github.com/apache/flink-connector-kafka/pull/314#discussion_r3980725781
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java:
##########
@@ -184,6 +188,33 @@ private void logFencedRequest(
}
}
+ /**
+ * With reused transactional ids, a fenced commit during recovery usually
means that the
+ * transaction was committed before the failure and its id was recycled
for a later checkpoint,
+ * whose transaction is still open on the broker under a newer epoch.
Nobody owns that
+ * transaction any more: the writer skips the id as precommitted and this
committer cannot
+ * commit it. Bumping the epoch aborts it, so that it does not block
read_committed consumers
+ * until the transaction timeout expires.
+ */
+ private void abortNewerTransaction(String transactionalId) {
+ FlinkKafkaInternalProducer<?, ?> producer =
+ producerFactory.apply(kafkaProducerConfig, transactionalId);
+ try {
+ producer.initTransactions();
+ LOG.info(
+ "Aborted open transaction of a newer epoch under {} after
its commit was fenced.",
+ transactionalId);
+ } catch (KafkaException e) {
+ LOG.warn(
+ "Could not abort the open transaction under {}; it stays
open until '{}' expires.",
+ transactionalId,
+ ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
+ e);
+ } finally {
+ producer.close();
+ }
+ }
Review Comment:
major: `catch (KafkaException)` only covers `initTransactions()`.
`producerFactory.apply(...)` (before the `try`) and `close()` (in `finally`)
can throw too, and since this runs inside the `ProducerFencedException` catch,
an escape skips `handleFailedTransaction` and `signalFailedWithKnownReason` —
best-effort cleanup failing the commit path. try-with-resources guards both the
acquisition and the close:
```suggestion
private void abortNewerTransaction(String transactionalId) {
try (FlinkKafkaInternalProducer<?, ?> producer =
producerFactory.apply(kafkaProducerConfig, transactionalId))
{
producer.initTransactions();
LOG.info(
"Bumped the epoch under {} to abort any lingering
transaction of a newer epoch after its commit was fenced.",
transactionalId);
} catch (KafkaException e) {
LOG.warn(
"Could not abort the open transaction under {}; it stays
open until '{}' expires.",
transactionalId,
ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
e);
}
}
```
Also reworded the `LOG.info` (init bumps the epoch even when nothing was
open, so "Aborted…" overclaims). Widen to `catch (RuntimeException)` if you
want it airtight against non-`KafkaException` throws.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
