ramu11 commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4423400130
##########
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 "
+ + "as one unit of work");
+ }
Review Comment:
addressed
##########
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:
addressed
--
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]
