gnodet-bot commented on code in PR #26557: URL: https://github.com/apache/camel/pull/26557#discussion_r4044803867
########## components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaTransactionSynchronizationTest.java: ########## @@ -0,0 +1,94 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.kafka; + +import java.util.Map; + +import org.apache.camel.Exchange; +import org.apache.camel.spi.UnitOfWork; +import org.apache.kafka.clients.consumer.ConsumerGroupMetadata; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.TopicPartition; +import org.junit.jupiter.api.Test; +import org.mockito.InOrder; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class KafkaTransactionSynchronizationTest { + + private static Exchange successfulExchange() { + Exchange exchange = mock(Exchange.class); + UnitOfWork uow = mock(UnitOfWork.class); + when(exchange.getUnitOfWork()).thenReturn(uow); + when(exchange.getException()).thenReturn(null); + when(exchange.isRollbackOnly()).thenReturn(false); + return exchange; + } + + @Test + void commitsConsumerOffsetsInsideTransactionForExactlyOnce() { + Producer<?, ?> producer = mock(Producer.class); + Map<TopicPartition, OffsetAndMetadata> offsets + = Map.of(new TopicPartition("orders", 0), new OffsetAndMetadata(43)); + ConsumerGroupMetadata groupMetadata = new ConsumerGroupMetadata("orders-group"); + + KafkaTransactionSynchronization sync + = new KafkaTransactionSynchronization("tx-1", producer, offsets, groupMetadata); + sync.onDone(successfulExchange()); + + // The offsets must be sent to the transaction before it is committed, so both happen atomically. + InOrder inOrder = inOrder(producer); + inOrder.verify(producer).sendOffsetsToTransaction(offsets, groupMetadata); + inOrder.verify(producer).commitTransaction(); + } + + @Test + void commitsWithoutSendingOffsetsWhenNotExactlyOnce() { + Producer<?, ?> producer = mock(Producer.class); + + KafkaTransactionSynchronization sync = new KafkaTransactionSynchronization("tx-1", producer); + sync.onDone(successfulExchange()); + + verify(producer).commitTransaction(); + verify(producer, never()).sendOffsetsToTransaction(any(), any()); + } + + @Test + void abortsTransactionWhenSendOffsetsFails() { + Producer<?, ?> producer = mock(Producer.class); + Map<TopicPartition, OffsetAndMetadata> offsets + = Map.of(new TopicPartition("orders", 0), new OffsetAndMetadata(43)); + ConsumerGroupMetadata groupMetadata = new ConsumerGroupMetadata("orders-group"); + doThrow(new KafkaException("boom")).when(producer).sendOffsetsToTransaction(offsets, groupMetadata); + + KafkaTransactionSynchronization sync + = new KafkaTransactionSynchronization("tx-1", producer, offsets, groupMetadata); + sync.onDone(successfulExchange()); + + // A failed sendOffsetsToTransaction must abort the (now open) transaction, not commit it. + verify(producer).abortTransaction(); + verify(producer, never()).commitTransaction(); Review Comment: ⚠️ **Test gap: `exchange.setException(e)` is not verified in the failure path.** The test confirms `abortTransaction()` is called and `commitTransaction()` is not, but doesn't verify that the exception is propagated to the exchange. If `exchange.setException(e)` were accidentally removed from `KafkaTransactionSynchronization`, this test would still pass — the failure would be silently swallowed from the caller's perspective. Add a `verify(exchange).setException(any(KafkaException.class))` assertion: ```suggestion verify(producer).abortTransaction(); verify(producer, never()).commitTransaction(); verify(exchange).setException(any(KafkaException.class)); } } ``` You'll also need `import static org.mockito.ArgumentMatchers.any;` (already imported) and to capture the exchange mock in the test so it can be verified — currently the mock is created inside `successfulExchange()` and not returned. Either return it from a helper or create it inline in this test method. ########## components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java: ########## @@ -529,13 +537,33 @@ private void startKafkaTransaction(Exchange exchange) { // 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)); + uow.addSynchronization(createTransactionSynchronization(exchange)); } else { LOG.debug("Using existing kafka transaction {} with exchange {}.", transactionId, exchange.getExchangeId()); } } + private KafkaTransactionSynchronization createTransactionSynchronization(Exchange exchange) { + if (configuration.isExactlyOnce()) { + KafkaManualCommit manual + = exchange.getMessage().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class); + if (manual instanceof DefaultKafkaManualCommit dmc) { + // Read the consumer group metadata on the consumer poll thread that is processing this exchange; the + // Kafka consumer is not safe for multi-threaded access. The offset to commit is the next offset to + // read, i.e. the processed record's offset + 1. + Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap( + dmc.getPartition(), new OffsetAndMetadata(dmc.getRecordOffset() + 1)); + return new KafkaTransactionSynchronization( + transactionId, kafkaProducer, offsets, dmc.getConsumerGroupMetadata()); + } + LOG.warn("exactlyOnce is enabled but no Kafka consumer manual-commit is present on the exchange; the source" + + " offsets will not be committed inside the transaction. Ensure the source Kafka consumer uses" + + " allowManualCommit=true and autoCommitEnable=false."); Review Comment: ⚠️ **Silent EOS degradation when a custom `KafkaManualCommit` is present.** The `instanceof DefaultKafkaManualCommit` check silently falls through for any custom `KafkaManualCommit` implementation (not extending `DefaultKafkaManualCommit`). The LOG.warn is emitted, but execution continues and returns a plain `KafkaTransactionSynchronization` — i.e., produces records without committing consumer offsets, violating the EOS guarantee the user explicitly requested with `exactlyOnce=true`. When `exactlyOnce=true` and a `KafkaManualCommit` is present but not the expected type, this should fail loudly rather than degrade silently: ```suggestion if (manual instanceof DefaultKafkaManualCommit dmc) { // Read the consumer group metadata on the consumer poll thread that is processing this exchange; the // Kafka consumer is not safe for multi-threaded access. The offset to commit is the next offset to // read, i.e. the processed record's offset + 1. Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap( dmc.getPartition(), new OffsetAndMetadata(dmc.getRecordOffset() + 1)); return new KafkaTransactionSynchronization( transactionId, kafkaProducer, offsets, dmc.getConsumerGroupMetadata()); } else if (manual != null) { // A custom KafkaManualCommit that doesn't extend DefaultKafkaManualCommit cannot supply // group metadata; failing loudly here is safer than silently producing without EOS. throw new IllegalStateException( "exactlyOnce=true requires a DefaultKafkaManualCommit instance to read consumer offsets; " + "found " + manual.getClass().getName() + ". Ensure the source consumer uses the default " + "KafkaManualCommitFactory."); } LOG.warn("exactlyOnce is enabled but no Kafka consumer manual-commit is present on the exchange; the source" + " offsets will not be committed inside the transaction. Ensure the source Kafka consumer uses" + " allowManualCommit=true and autoCommitEnable=false."); ``` The no-header case (manual == null) can remain a warning since it could be a non-Kafka source intentionally chained before the Kafka producer. -- 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]
