zhfeng commented on code in PR #7454:
URL: https://github.com/apache/camel/pull/7454#discussion_r857385871
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -492,4 +509,40 @@ private void doSend(Object key, ProducerRecord<Object,
Object> record, KafkaProd
kafkaProducer.send(record, cb);
}
}
+
+ private void startKafkaTransaction(Exchange exchange) {
+ exchange.getUnitOfWork().beginTransactedBy(transactionId);
+ kafkaProducer.beginTransaction();
+ exchange.getUnitOfWork().addSynchronization(new
KafkaTransactionSynchronization(transactionId, kafkaProducer));
+ }
+}
+
+class KafkaTransactionSynchronization extends SynchronizationAdapter {
+ private static final Logger LOG =
LoggerFactory.getLogger(KafkaTransactionSynchronization.class);
+ private String transactionId;
+ private Producer kafkaProducer;
+
+ public KafkaTransactionSynchronization(String transacionId, Producer
kafkaProducer) {
+ this.transactionId = transacionId;
+ this.kafkaProducer = kafkaProducer;
+ }
+
+ @Override
+ public void onDone(Exchange exchange) {
+ try {
+ if (exchange.getException() != null || exchange.isRollbackOnly()) {
Review Comment:
It seems ```ProducerFencedException``` should be concerned
>This fatal exception indicates that another producer with the same
<code>transactional.id</code> has been
started. It is only possible to have one producer instance with a
<code>transactional.id</code> at any
given time, and the latest one to be started "fences" the previous
instances so that they can no longer
make transactional requests. When you encounter this exception, you must
close the producer instance.
I'm not familiar with kafka. So @alexkazan87 can you check it?
--
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]
