gnodet-bot commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4430599962
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaTransactionSynchronization.java:
##########
@@ -28,53 +34,154 @@
class KafkaTransactionSynchronization extends SynchronizationAdapter {
private static final Logger LOG =
LoggerFactory.getLogger(KafkaTransactionSynchronization.class);
private final String transactionId;
- private final Producer kafkaProducer;
+ private Producer kafkaProducer;
private final KafkaProducer owner;
+ private final Map<TopicPartition, OffsetAndMetadata> offsetsToCommit;
+ private final ConsumerGroupMetadata groupMetadata;
+ private final String ownerExchangeId;
public KafkaTransactionSynchronization(String transactionId, Producer
kafkaProducer, KafkaProducer owner) {
+ this(transactionId, kafkaProducer, owner, null, null, null);
+ }
+
+ public KafkaTransactionSynchronization(
+ String transactionId, Producer
kafkaProducer, KafkaProducer owner,
+ Map<TopicPartition,
OffsetAndMetadata> offsetsToCommit,
+ ConsumerGroupMetadata
groupMetadata) {
+ this(transactionId, kafkaProducer, owner, offsetsToCommit,
groupMetadata, null);
+ }
+
+ public KafkaTransactionSynchronization(
+ String transactionId, Producer
kafkaProducer, KafkaProducer owner,
+ Map<TopicPartition,
OffsetAndMetadata> offsetsToCommit,
+ ConsumerGroupMetadata groupMetadata,
+ String ownerExchangeId) {
this.transactionId = transactionId;
this.kafkaProducer = kafkaProducer;
this.owner = owner;
+ this.offsetsToCommit = offsetsToCommit;
+ this.groupMetadata = groupMetadata;
+ this.ownerExchangeId = ownerExchangeId;
+ }
+
+ public KafkaTransactionSynchronization(
+ String transactionId, Producer
kafkaProducer, KafkaProducer owner,
+ KafkaExactlyOnceMetadata metadata,
String ownerExchangeId) {
+ this(transactionId, kafkaProducer, owner,
+ metadata == null ? null : metadata.getOffsetsToCommit(),
+ metadata == null ? null : metadata.getConsumerGroupMetadata(),
ownerExchangeId);
+ }
+
+ /** Keep exactly-once synchronization bound to the source exchange; retain
legacy handover for ordinary sends. */
+ @Override
+ public boolean allowHandover() {
+ return offsetsToCommit == null;
+ }
+
+ /** Remember send failures even if a Camel error handler clears the
exchange exception later. */
+ public static void recordSendFailure(Exchange exchange, Throwable failure)
{
+ KafkaExactlyOnceMetadata.recordSendFailure(exchange, failure);
}
@Override
public void onDone(Exchange exchange) {
try {
+ Throwable sendFailure =
exchange.getProperty(KafkaExactlyOnceMetadata.SEND_FAILURE_PROPERTY,
Throwable.class);
if (exchange.getException() != null || exchange.isRollbackOnly()) {
rollback(exchange);
+ markAborted(exchange);
+ } else if (offsetsToCommit != null && sendFailure != null) {
+ // The route error handler handled a failed Kafka send. Abort
every output record from the failed
+ // transaction, then commit only the source offsets in a fresh
transaction so the poison record is
+ // deliberately skipped instead of being replayed forever.
+ commitOffsetsAfterHandledSendFailure(exchange, sendFailure);
} else {
commit(exchange);
+ if (exchange.getException() != null) {
+ markAborted(exchange);
+ }
}
} finally {
- exchange.getUnitOfWork().endTransactedBy(transactionId);
+ try {
+ exchange.getUnitOfWork().endTransactedBy(transactionId);
+ } finally {
+ if (offsetsToCommit != null) {
+ // allowHandover() is false, but keep the acquired ID as
the source of truth even if a caller
+ // completes a copied exchange.
+ owner.releaseExactlyOnceTransaction(
+ ownerExchangeId != null ? ownerExchangeId :
exchange.getExchangeId());
+ }
+ }
+ }
+ }
+
+ private void markAborted(Exchange exchange) {
+ if (offsetsToCommit != null) {
+
exchange.setProperty(KafkaExactlyOnceMetadata.TRANSACTION_ABORTED_PROPERTY,
Boolean.TRUE);
}
}
private void commit(Exchange exchange) {
try {
LOG.debug("Commit kafka transaction {} with exchange {}",
transactionId, exchange.getExchangeId());
+ if (offsetsToCommit != null) {
+ LOG.debug("Sending {} consumer offset(s) to kafka transaction
{}",
+ offsetsToCommit.size(), transactionId);
+ kafkaProducer.sendOffsetsToTransaction(offsetsToCommit,
groupMetadata);
+ }
kafkaProducer.commitTransaction();
} catch (Exception e) {
- // The commit failed. Record it and return the producer to a
usable state - abort the still-open
- // transaction, or close and rebuild it when the error is fatal.
The previous code only recorded the
- // exception, leaving the transaction open so the next
beginTransaction() failed and wedged the route
- // (CAMEL-24782).
+ if (offsetsToCommit != null) {
+ LOG.warn("Kafka exactly-once offset commit failed. Verify that
the source consumer and destination "
+ + "producer use the same Kafka cluster and that the
captured consumer group generation "
+ + "is current.",
+ e);
+ }
exchange.setException(e);
recover(e, "commit");
}
}
+ private void commitOffsetsAfterHandledSendFailure(Exchange exchange,
Throwable sendFailure) {
+ try {
+ kafkaProducer.abortTransaction();
+ } catch (Exception abortFailure) {
+ // The old producer may be fenced or left in an unusable
transaction state. Close it and
+ // initialize a fresh producer with the same transactional.id
before trying the offset-only
+ // transaction. If recreation fails, leave the source record
uncommitted so the consumer replays it.
+ try {
+ closeForRecreation();
+ owner.recreateProducerIfClosed();
+ kafkaProducer = owner.getKafkaProducer();
+ exchange.setException(null);
Review Comment:
Nit: `exchange.setException(null)` is a defensive no-op here — the exchange
enters `commitOffsetsAfterHandledSendFailure` with `exception == null` (the
`onDone` guard at the top of the method ensures this). The intent is presumably
to clear any transient state set inside the abort-failure `catch` block before
proceeding to the offset-only transaction, but nothing in that `catch` path
sets an exception on the exchange before reaching this line (the recreation
failure branch returns early). The call is harmless, and the test explicitly
verifies it; consider replacing it with a comment explaining the invariant
instead.
--
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]
