This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new ad9645a6237a CAMEL-25908: camel-knative-http - a consumer started
again receives the requests of its path again (#28432)
ad9645a6237a is described below
commit ad9645a6237a793788b3791c8751370e099b8ba1
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 25 13:01:31 2026 +0530
CAMEL-25908: camel-knative-http - a consumer started again receives the
requests of its path again (#28432)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../knative/http/KnativeHttpConsumer.java | 2 +
.../component/knative/http/KnativeHttpTest.java | 82 ++++++++++++++++++++++
2 files changed, 84 insertions(+)
diff --git
a/components/camel-knative/camel-knative-http/src/main/java/org/apache/camel/component/knative/http/KnativeHttpConsumer.java
b/components/camel-knative/camel-knative-http/src/main/java/org/apache/camel/component/knative/http/KnativeHttpConsumer.java
index 651207e928eb..0608e48c438c 100644
---
a/components/camel-knative/camel-knative-http/src/main/java/org/apache/camel/component/knative/http/KnativeHttpConsumer.java
+++
b/components/camel-knative/camel-knative-http/src/main/java/org/apache/camel/component/knative/http/KnativeHttpConsumer.java
@@ -193,7 +193,9 @@ public class KnativeHttpConsumer extends DefaultConsumer {
@Override
protected void doStop() throws Exception {
if (route != null) {
+ // a removed route cannot be added to the router again, so a start
creates a new one
route.remove();
+ route = null;
}
super.doStop();
diff --git
a/components/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
b/components/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
index 1c63aede2ac0..50dcb5a6f757 100644
---
a/components/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
+++
b/components/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
@@ -31,6 +31,7 @@ import io.vertx.core.http.HttpServerRequest;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelException;
import org.apache.camel.CamelExecutionException;
+import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
import org.apache.camel.FailedToStartRouteException;
import org.apache.camel.ProducerTemplate;
@@ -44,9 +45,11 @@ import org.apache.camel.component.knative.spi.Knative;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.http.base.HttpOperationFailedException;
import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.model.RouteDefinition;
import org.apache.camel.support.PropertyBindingSupport;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.throttling.ThrottlingExceptionRoutePolicy;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StringHelper;
import org.assertj.core.api.Assertions;
@@ -66,6 +69,7 @@ import static
org.apache.camel.component.knative.http.KnativeHttpTestSupport.con
import static
org.apache.camel.component.knative.http.KnativeHttpTestSupport.httpAttribute;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.awaitility.Awaitility.await;
import static org.hamcrest.Matchers.emptyOrNullString;
import static org.hamcrest.Matchers.is;
@@ -986,6 +990,84 @@ public class KnativeHttpTest {
});
}
+ @ParameterizedTest
+ @EnumSource(CloudEvents.class)
+ void testConsumerStoppedAndStartedAgain(CloudEvent ce) throws Exception {
+ configureTwoConsumersAndStart(ce, r2 -> {
+ });
+
+ // the consumer is not Suspendable, so a route policy that suspends
and resumes it stops and starts it again
+ Consumer consumer = context.getRoute("r2").getConsumer();
+ ServiceHelper.stopService(consumer);
+ assertThat(template.request("direct:start", e ->
e.getMessage().setBody("h2"))).satisfies(e -> {
+ assertThat(e.isFailed()).isTrue();
+
assertThat(e.getException()).isInstanceOf(HttpOperationFailedException.class);
+ });
+
+ ServiceHelper.startService(consumer);
+
+ assertThat(template.requestBody("direct:start", "h2",
String.class)).isEqualTo("r2");
+ }
+
+ @ParameterizedTest
+ @EnumSource(CloudEvents.class)
+ void testConsumerStartedAgainByCircuitBreakerRoutePolicy(CloudEvent ce)
throws Exception {
+ ThrottlingExceptionRoutePolicy policy = new
ThrottlingExceptionRoutePolicy(1, 60000, 1000, null);
+ AtomicBoolean fail = new AtomicBoolean(true);
+ configureTwoConsumersAndStart(ce, r2 ->
r2.routePolicy(policy).process(e -> {
+ if (fail.get()) {
+ throw new IllegalStateException("Forced");
+ }
+ }));
+ Consumer consumer = context.getRoute("r2").getConsumer();
+
+ // the failure opens the circuit, which stops the consumer as it is
not Suspendable
+ assertThat(template.request("direct:start", e ->
e.getMessage().setBody("h2")).isFailed()).isTrue();
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
ServiceHelper.isStopped(consumer));
+ fail.set(false);
+
+ // after halfOpenAfter the circuit is half open: the consumer is
started again to let a request through
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
ServiceHelper.isStarted(consumer));
+
+ assertThat(template.requestBody("direct:start", "h2",
String.class)).isEqualTo("r2");
+ }
+
+ private void configureTwoConsumersAndStart(CloudEvent ce,
java.util.function.Consumer<RouteDefinition> r2)
+ throws Exception {
+ configureKnativeComponent(
+ context,
+ ce,
+ sourceEndpoint(
+ "ep1",
+ Map.of(
+ Knative.KNATIVE_CLOUD_EVENT_TYPE,
"org.apache.camel.event",
+ Knative.CONTENT_TYPE, "text/plain",
+ Knative.KNATIVE_FILTER_PREFIX + "h", "h1")),
+ sourceEndpoint(
+ "ep2",
+ Map.of(
+ Knative.KNATIVE_CLOUD_EVENT_TYPE,
"org.apache.camel.event",
+ Knative.CONTENT_TYPE, "text/plain",
+ Knative.KNATIVE_FILTER_PREFIX + "h", "h2")));
+
+ RouteBuilder.addRoutes(context, b -> {
+ b.from("knative:endpoint/ep1")
+ .routeId("r1")
+ .setBody().simple("${routeId}");
+ RouteDefinition route = b.from("knative:endpoint/ep2")
+ .routeId("r2");
+ r2.accept(route);
+ route.setBody().simple("${routeId}");
+ });
+ RouteBuilder.addRoutes(context, b -> {
+ b.from("direct:start")
+ .setHeader("h").body()
+ .toF("http://localhost:%d";, platformHttpPort);
+ });
+
+ context.start();
+ }
+
@ParameterizedTest
@EnumSource(CloudEvents.class)
void testAddConsumer(CloudEvent ce) throws Exception {
