davsclaus commented on PR #25828:
URL: https://github.com/apache/camel/pull/25828#issuecomment-5460963136
CI is failing on `KafkaSagaIT.testSaga` (both `build (25, false)` and `build
(17, false)` — the latter was just cancelled by fail-fast after the former
failed), not because of flakiness but as a direct consequence of this PR's own
change:
```
java.lang.AssertionError: mock://result Received message count. Expected:
<1> but was: <0>
at
org.apache.camel.component.kafka.integration.KafkaSagaIT.testSaga(KafkaSagaIT.java:39)
```
`KafkaSagaIT` starts a saga on `direct:saga` with a plain
`InMemorySagaService`, then routes the exchange through a real Kafka round-trip
to a second route that does `.saga().propagation(MANDATORY)`. The saga id can
only survive that hop via the `Long-Running-Action` header — the
exchange-extension internal state doesn't carry across the new `Exchange` the
Kafka consumer creates. That's exactly the header fallback this PR now gates
behind `isLongRunningActionHeaderSupported()`, which `InMemorySagaService`
correctly defaults to `false`, so the consumer-side `MANDATORY` step can no
longer find a coordinator to join.
This matches the migration path your own upgrade-guide entry documents: *"A
custom `CamelSagaService` that relies on the header to join sagas started by
another participant must override the new method."* `KafkaSagaIT`'s scenario is
exactly that case, it just wasn't updated in this PR.
I confirmed this is the only test affected by the change (checked all other
`.saga()` usages across core/components — the rest route via `direct:`/`seda:`
in the same JVM, where the exchange-extension state already survives, and the
`camel-lra` tests already use `LRASagaService`, which overrides the new method).
Suggested fix — verified locally (`mvn verify -Dit.test=KafkaSagaIT` passes:
`Tests run: 1, Failures: 0`):
```diff
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaSagaIT.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaSagaIT.java
@@ -45,7 +45,7 @@ public class KafkaSagaIT extends BaseKafkaTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- getCamelContext().addService(new InMemorySagaService());
+ getCamelContext().addService(new KafkaInteropSagaService());
from("direct:saga")
.saga()
@@ -64,6 +64,20 @@ public class KafkaSagaIT extends BaseKafkaTestSupport {
}
}
+/**
+ * The saga id is stored in the exchange's internal state, which does not
survive the Kafka produce/consume round-trip -
+ * only the {@code Long-Running-Action} header does. Advertise header
support so the consumer route can join the saga
+ * started by the producer route, as documented for a {@code
CamelSagaService} that relies on the header to join sagas
+ * started by another participant.
+ */
+final class KafkaInteropSagaService extends InMemorySagaService {
+
+ @Override
+ public boolean isLongRunningActionHeaderSupported() {
+ return true;
+ }
+}
+
final class SagaBean {
public static String id;
public static Boolean isSame = false;
```
_Claude Code on behalf of davsclaus_
--
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]
