gnodet-bot commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4422833043
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -568,23 +581,146 @@ private void startKafkaTransaction(Exchange exchange) {
UnitOfWork uow = exchange.getUnitOfWork();
if (!uow.isTransactedBy(transactionId)) {
- // A prior transaction may have hit a fatal error that closed the
shared producer; rebuild it before
- // starting a new transaction so the route recovers instead of
staying wedged on a dead producer
- // (CAMEL-24782).
- recreateProducerIfClosed();
- LOG.debug("Starting kafka transaction {} with exchange {}",
transactionId, exchange.getExchangeId());
- // Begin the broker transaction first, then mark the unit of work
and register the
- // synchronization. This way a failure in beginTransaction() does
not leave the unit of work
- // flagged as transacted without a synchronization to commit or
roll it back (CAMEL-24780).
- kafkaProducer.beginTransaction();
- uow.beginTransactedBy(transactionId);
- uow.addSynchronization(new
KafkaTransactionSynchronization(transactionId, kafkaProducer, this));
+ // Resolve and validate source metadata before opening the broker
transaction. A failure here must not
+ // leave an open transaction without a registered synchronization.
+ KafkaExactlyOnceMetadata exactlyOnceMetadata
+ = configuration.isExactlyOnce() ?
createExactlyOnceMetadata(exchange) : null;
+ if (exactlyOnceMetadata != null) {
+ String registeredProducer =
exchange.getProperty(EXACTLY_ONCE_PRODUCER_PROPERTY, String.class);
+ if (registeredProducer != null &&
!registeredProducer.equals(transactionId)) {
+ throw new IllegalStateException(
+ "Only one exactlyOnce Kafka producer is supported
per exchange/unit of work");
+ }
+ if (exchange.getProperty(Exchange.SPLIT_INDEX) != null
+ || exchange.getProperty(Exchange.MULTICAST_INDEX) !=
null) {
+ throw new IllegalStateException(
+ "exactlyOnce does not support split/multicast
child exchanges; process the source record "
Review Comment:
💡 **Dead code:** This split/multicast guard is unreachable.
`createExactlyOnceMetadata(exchange)` is called at line 587, before this block,
and contains the identical check (it throws there first). Remove this duplicate.
```suggestion
```
--
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]
