You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "oscerd (via GitHub)" <gi...@apache.org> on 2026/09/16 16:24:15 UTC
[PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
oscerd opened a new pull request, #26523:
URL: https://github.com/apache/camel/pull/26523
## What this does
Fixes two related robustness problems in the **transactional producer** async path.
### 1. Transaction begin (and body resolution) ran outside the `try`
`KafkaProducer.process(Exchange, AsyncCallback)` resolved `message.getBody()` and called `startKafkaTransaction(exchange)` **before** the `try` that guards sending. If `kafkaProducer.beginTransaction()` threw (or a lazy body conversion threw), the exception escaped `process()` **without calling `callback.done(...)`**, violating the async producer contract (the caller's callback is never completed).
**Fix:** the body resolution and `startKafkaTransaction` call are now inside the `try`, so a failure is handled like any other send failure β the exception is set on the exchange and the async callback is completed.
### 2. UnitOfWork marked transacted before the transaction actually began
In `startKafkaTransaction` the order was `uow.beginTransactedBy(id)` β `kafkaProducer.beginTransaction()` β `uow.addSynchronization(...)`. If `beginTransaction()` threw, the UoW was already flagged as transacted by this id but **no synchronization was registered**, so nothing would ever commit/abort it and the flag dangled for the rest of the exchange.
**Fix:** call `kafkaProducer.beginTransaction()` first, and only then mark the UoW and register the synchronization, so a begin failure leaves the UoW untouched.
Both changes are limited to the transactional async path; non-transactional sends are unaffected.
## Tests
- `KafkaProducerTest#processAsyncCompletesCallbackWhenBeginTransactionFails` β makes `beginTransaction()` throw and asserts that `process()` sets the exception and completes the async callback (instead of throwing), and that the UnitOfWork is **not** marked transacted and no synchronization is registered. This test fails on the previous code (the exception escaped `process()` and the UoW was already flagged).
Full `camel-kafka` module build (unit + integration tests) is green.
JIRA: https://issues.apache.org/jira/browse/CAMEL-24780
_Claude Code on behalf of @oscerd_
π€ Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd closed pull request #26523: CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback
URL: https://github.com/apache/camel/pull/26523
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on PR #26523:
URL: https://github.com/apache/camel/pull/26523#issuecomment-5701055100
**Backport needed.**
The same bug exists in both active maintenance branches:
- `camel-4.22.x` β `process(Exchange, AsyncCallback)` lines 479-485: `body` resolved and `startKafkaTransaction()` called outside the `try` block; `startKafkaTransaction()` has the same UoW-before-broker order
- `camel-4.18.x` β same pattern
This is a correctness bug (missed async callback + dangling UoW transacted flag on begin failure), not a behaviour change, so backporting to both LTS branches seems warranted.
_This comment was generated by an AI agent, Hermès on behalf of @gnodet._
<!-- reviewer: gnodet-bot -->
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "gnodet (via GitHub)" <gi...@apache.org>.
gnodet commented on PR #26523:
URL: https://github.com/apache/camel/pull/26523#issuecomment-5717294684
<!-- camel-backport-bot -->
### π Backport Bot
**Port PRs created:**
- β
`camel-4.22.x` β #26555
- β
`camel-4.18.x` β #26556
βΉοΈ If you push additional commits after `/port`, use `/port` again to update the port PRs.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26523:
URL: https://github.com/apache/camel/pull/26523#issuecomment-5700927296
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd commented on code in PR #26523:
URL: https://github.com/apache/camel/pull/26523#discussion_r4035441951
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java:
##########
@@ -189,6 +192,39 @@ public void processAsyncSendsMessageWithException() {
assertRecordMetadataExists();
}
+ @Test
+ public void processAsyncCompletesCallbackWhenBeginTransactionFails() throws Exception {
+ // CAMEL-24780: a failure to begin the transaction must set the exception and complete the async
+ // callback rather than escaping process(), and it must not leave the unit of work flagged as
+ // transacted without a synchronization to commit or roll it back.
+ setTransactionId(producer, "test-tx");
Review Comment:
Applied in `ae932f8` β replaced the reflection with `endpoint.getConfiguration().setTransactionalId("test-tx")` + `producer.doStart()`, so `transactionId` is derived through the real init path. As you noted, `doStart()` doesn't recreate the producer (the mock is already set via `setKafkaProducer`), `initTransactions()` is a no-op on the mock, and the worker pool is pre-set β so the test stays self-contained and the `setTransactionId` helper + `java.lang.reflect.Field` import are gone. Thanks for the cleaner approach!
_Claude Code on behalf of @oscerd_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "gnodet (via GitHub)" <gi...@apache.org>.
gnodet commented on PR #26523:
URL: https://github.com/apache/camel/pull/26523#issuecomment-5716779188
<!-- camel-backport-bot -->
### π Backport Bot
This bugfix targets `main` and may need porting to:
- `camel-4.22.x`
Labels `port/<branch>` have been added. Remove a label to opt out of porting to that branch.
Port PRs will be created automatically when this PR is merged. Comment `/port` to create them now.
βΉοΈ If you push additional commits after `/port`, use `/port` again to update the port PRs.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #26523:
URL: https://github.com/apache/camel/pull/26523#discussion_r4028430702
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java:
##########
@@ -189,6 +192,39 @@ public void processAsyncSendsMessageWithException() {
assertRecordMetadataExists();
}
+ @Test
+ public void processAsyncCompletesCallbackWhenBeginTransactionFails() throws Exception {
+ // CAMEL-24780: a failure to begin the transaction must set the exception and complete the async
+ // callback rather than escaping process(), and it must not leave the unit of work flagged as
+ // transacted without a synchronization to commit or roll it back.
+ setTransactionId(producer, "test-tx");
+ endpoint.getConfiguration().setTopic("sometopic");
+
+ Producer kp = producer.getKafkaProducer();
+ Mockito.doThrow(new ApiException("cannot begin")).when(kp).beginTransaction();
+
+ UnitOfWork uow = Mockito.mock(UnitOfWork.class);
+ Mockito.when(uow.isTransactedBy(any())).thenReturn(false);
+ Mockito.when(exchange.getUnitOfWork()).thenReturn(uow);
+ Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getMessage()).thenReturn(in);
+
+ boolean sync = producer.process(exchange, callback);
+
+ assertTrue(sync);
+ Mockito.verify(exchange).setException(isA(ApiException.class));
+ Mockito.verify(callback).done(eq(true));
+ // begin failed, so the unit of work must be left untouched (no dangling transacted flag)
+ Mockito.verify(uow, Mockito.never()).beginTransactedBy(any());
+ Mockito.verify(uow, Mockito.never()).addSynchronization(any());
+ }
Review Comment:
**Missing test for the "transaction started, send fails" path.**
The new test covers `beginTransaction()` throwing β good. But the other half of Fix 1's correctness story is: when `beginTransaction()` **succeeds** and then `doSend()` / `createRecord()` fails, the `KafkaTransactionSynchronization` that was registered must fire `abortTransaction()` via the UoW lifecycle.
This path is handled by the existing Camel UoW mechanism (not by any code introduced here), but a test asserting that `kafkaProducer.abortTransaction()` is called when a send fails in transactional mode would provide a regression guard for the full scenario and make the contract explicit.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -518,8 +521,11 @@ private void startKafkaTransaction(Exchange exchange) {
if (!uow.isTransactedBy(transactionId)) {
LOG.debug("Starting kafka transaction {} with exchange {}", transactionId, exchange.getExchangeId());
- uow.beginTransactedBy(transactionId);
+ // 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));
Review Comment:
**Latent window between `beginTransaction()` and `addSynchronization()`.**
The reorder is correct for the `beginTransaction()` failure case. However, if `beginTransaction()` succeeds and then anything between lines 528-529 threw (e.g. an `OutOfMemoryError` or a misbehaving `UnitOfWork` implementation), the Kafka transaction would be open with no synchronization registered to abort it.
In practice `DefaultUnitOfWork.beginTransactedBy()` and `addSynchronization()` are in-memory ArrayList ops that won't throw, so this is theoretical. But guarding the 3-step sequence makes the intent explicit and future-proof:
```suggestion
kafkaProducer.beginTransaction();
try {
uow.beginTransactedBy(transactionId);
uow.addSynchronization(new KafkaTransactionSynchronization(transactionId, kafkaProducer));
} catch (Exception uowEx) {
// beginTransaction() succeeded but UoW bookkeeping failed β abort the open transaction
// to avoid a dangling half-begun transaction on the broker.
try {
kafkaProducer.abortTransaction();
} catch (Exception abortEx) {
uowEx.addSuppressed(abortEx);
}
throw uowEx;
}
```
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26523:
URL: https://github.com/apache/camel/pull/26523#issuecomment-5701947460
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `components/camel-kafka`
---
<details><summary>:microscope: Scalpel shadow comparison β Scalpel: 9 tested, 27 compile-only β current: 9 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **36 affected modules** (current approach: 9).
<details><summary>:warning: Modules only in Scalpel (27)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `docs`
- `dummy-component`
</details>
**Skip-tests mode would test 9 modules** (1 direct + 8 downstream), **skip tests for 27** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (9)</summary>
- `camel-jbang-mcp`
- `camel-jbang-plugin-mcp`
- `camel-jbang-plugin-route-parser`
- `camel-jbang-plugin-tui`
- `camel-jbang-plugin-validate`
- `camel-kafka`
- `camel-launcher-container`
- `camel-yaml-dsl-validator`
- `camel-yaml-dsl-validator-maven-plugin`
</details>
<details><summary>Modules with tests skipped (27)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `docs`
- `dummy-component`
</details>
> :information_source: Shadow mode β Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
:warning: **Some tests are disabled on GitHub Actions** (`@DisabledIfSystemProperty(named = "ci.env.name")`) and require manual verification:
- `components/camel-kafka`: 2 test(s) disabled on GitHub Actions
<details><summary>All tested modules (36 modules, 4m 30s total)</summary>
**Total reactor time:** 4m 30s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Launcher | 51.1s | SUCCESS |
| Camel :: JBang :: MCP | 37.3s | SUCCESS |
| Camel :: JBang :: Plugin :: TUI | 28.9s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 22.8s | SUCCESS |
| Camel :: Component DSL | 20.7s | SUCCESS |
| Camel :: Docs | 16.9s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 13.1s | SUCCESS |
| Camel :: Kamelet Main | 8.3s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 8.0s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 7.4s | SUCCESS |
| Camel :: All Components Sync point | 7.3s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 7.1s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 5.7s | SUCCESS |
| Camel :: YAML DSL :: Validator | 5.1s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 5.0s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 3.6s | SUCCESS |
| Camel :: Catalog :: Maven | 3.0s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 2.8s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 2.1s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 2.1s | SUCCESS |
| Camel :: Coverage | 2.0s | SUCCESS |
| Camel :: Assembly | 1.8s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 1.5s | SUCCESS |
| Camel :: JBang :: Main | 1.0s | SUCCESS |
| Camel :: JBang :: Integration tests | 1.0s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 1.0s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 0.9s | SUCCESS |
| Camel :: Catalog :: Console | 0.9s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 0.8s | SUCCESS |
| Camel :: JBang :: Plugin :: Route Parser | 0.8s | SUCCESS |
| Camel :: Launcher :: Container | 0.7s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
| Camel :: Kafka | n/a | |
| Camel :: YAML DSL | n/a | |
**Top 20 slowest modules:**
- `Camel :: Launcher` (51.1s)
- `Camel :: JBang :: MCP` (37.3s)
- `Camel :: JBang :: Plugin :: TUI` (28.9s)
- `Camel :: Catalog :: Camel Catalog` (22.8s)
- `Camel :: Component DSL` (20.7s)
- `Camel :: Docs` (16.9s)
- `Camel :: JBang :: Plugin :: Kubernetes` (13.1s)
- `Camel :: Kamelet Main` (8.3s)
- `Camel :: Catalog :: Camel Route Parser` (8.0s)
- `Camel :: JBang :: Plugin :: Testing` (7.4s)
- `Camel :: All Components Sync point` (7.3s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (7.1s)
- `Camel :: YAML DSL :: Deserializers` (5.7s)
- `Camel :: YAML DSL :: Validator` (5.1s)
- `Camel :: JBang :: Plugin :: Validate` (5.0s)
- `Camel :: YAML DSL :: Maven Plugins` (3.6s)
- `Camel :: Catalog :: Maven` (3.0s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (2.8s)
- `Camel :: Catalog :: Suggest (deprecated)` (2.1s)
- `Camel :: Catalog :: Dummy Component` (2.1s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/35121672767)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd commented on code in PR #26523:
URL: https://github.com/apache/camel/pull/26523#discussion_r4029169975
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java:
##########
@@ -518,8 +521,11 @@ private void startKafkaTransaction(Exchange exchange) {
if (!uow.isTransactedBy(transactionId)) {
LOG.debug("Starting kafka transaction {} with exchange {}", transactionId, exchange.getExchangeId());
- uow.beginTransactedBy(transactionId);
+ // 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));
Review Comment:
Thanks β I'm leaving this as-is. The window is between two in-memory operations (`beginTransactedBy` adds to a set, `addSynchronization` adds to a list) that don't throw under normal operation; the only realistic trigger is a JVM-level error such as `OutOfMemoryError`, in which case the follow-up `abortTransaction()` broker call would almost certainly fail too. Wrapping it adds nesting for a path that can't meaningfully recover, so I'd rather keep the three-step sequence readable.
_Claude Code on behalf of @oscerd_
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java:
##########
@@ -189,6 +192,39 @@ public void processAsyncSendsMessageWithException() {
assertRecordMetadataExists();
}
+ @Test
+ public void processAsyncCompletesCallbackWhenBeginTransactionFails() throws Exception {
+ // CAMEL-24780: a failure to begin the transaction must set the exception and complete the async
+ // callback rather than escaping process(), and it must not leave the unit of work flagged as
+ // transacted without a synchronization to commit or roll it back.
+ setTransactionId(producer, "test-tx");
+ endpoint.getConfiguration().setTopic("sometopic");
+
+ Producer kp = producer.getKafkaProducer();
+ Mockito.doThrow(new ApiException("cannot begin")).when(kp).beginTransaction();
+
+ UnitOfWork uow = Mockito.mock(UnitOfWork.class);
+ Mockito.when(uow.isTransactedBy(any())).thenReturn(false);
+ Mockito.when(exchange.getUnitOfWork()).thenReturn(uow);
+ Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getMessage()).thenReturn(in);
+
+ boolean sync = producer.process(exchange, callback);
+
+ assertTrue(sync);
+ Mockito.verify(exchange).setException(isA(ApiException.class));
+ Mockito.verify(callback).done(eq(true));
+ // begin failed, so the unit of work must be left untouched (no dangling transacted flag)
+ Mockito.verify(uow, Mockito.never()).beginTransactedBy(any());
+ Mockito.verify(uow, Mockito.never()).addSynchronization(any());
+ }
Review Comment:
The abort-on-send-failure path is handled by the existing Camel `UnitOfWork` lifecycle (the registered `KafkaTransactionSynchronization.onDone` β `abortTransaction()`), which this PR does not change β it's exercised by the transactional integration tests. Since it's pre-existing behavior rather than something introduced here, I've kept this PR focused on the begin-ordering fix. Happy to add dedicated `KafkaTransactionSynchronization` unit tests in a follow-up if that's useful.
_Claude Code on behalf of @oscerd_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #26523:
URL: https://github.com/apache/camel/pull/26523#discussion_r4029567213
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java:
##########
@@ -189,6 +192,39 @@ public void processAsyncSendsMessageWithException() {
assertRecordMetadataExists();
}
+ @Test
+ public void processAsyncCompletesCallbackWhenBeginTransactionFails() throws Exception {
+ // CAMEL-24780: a failure to begin the transaction must set the exception and complete the async
+ // callback rather than escaping process(), and it must not leave the unit of work flagged as
+ // transacted without a synchronization to commit or roll it back.
+ setTransactionId(producer, "test-tx");
Review Comment:
Optional: the reflection is needed only because the fixture never calls `doStart()`, where `transactionId` is derived. `endpoint.getConfiguration().setTransactionalId("test-tx"); producer.doStart();` would exercise the real init path (`initTransactions()` is a no-op on the mock and the worker pool is already set), and the `setTransactionId` helper could go. Fine as is if you prefer to keep the test independent of `doStart()`.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd commented on PR #26523:
URL: https://github.com/apache/camel/pull/26523#issuecomment-5702153633
**Backports:** added the `port/camel-4.22.x` and `port/camel-4.18.x` labels β the same pre-try transaction start exists on both maintenance branches; backport PRs will follow once this merges on `main`. Replied inline to the two code suggestions.
_Claude Code on behalf of @oscerd_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24780: camel-kafka - fix transactional producer leaking a half-begun transaction and skipping the async callback [camel]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd merged PR #26523:
URL: https://github.com/apache/camel/pull/26523
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org