diff --git a/components/camel-knative/camel-knative-component/src/main/java/org/apache/camel/component/knative/ce/AbstractCloudEventProcessor.java b/components/camel-knative/camel-knative-component/src/main/java/org/apache/camel/component/knative/ce/AbstractCloudEventProcessor.java index 0e7f258d33830..05d52e48d895a 100644 --- a/components/camel-knative/camel-knative-component/src/main/java/org/apache/camel/component/knative/ce/AbstractCloudEventProcessor.java +++ b/components/camel-knative/camel-knative-component/src/main/java/org/apache/camel/component/knative/ce/AbstractCloudEventProcessor.java @@ -19,16 +19,20 @@ import java.io.InputStream; import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; +import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.function.Supplier; import org.apache.camel.Exchange; +import org.apache.camel.Message; import org.apache.camel.Processor; import org.apache.camel.cloudevents.CloudEvent; import org.apache.camel.component.knative.KnativeEndpoint; import org.apache.camel.component.knative.spi.Knative; import org.apache.camel.component.knative.spi.KnativeResource; +import org.apache.camel.spi.HeaderFilterStrategy; +import org.apache.camel.support.DefaultHeaderFilterStrategy; import org.apache.camel.util.StringHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,6 +40,10 @@ abstract class AbstractCloudEventProcessor implements CloudEventProcessor { private final CloudEvent cloudEvent; + // Filters internal Camel headers (Camel*/camel*) from structured-mode CloudEvent extensions, keeping the + // extension-to-header mapping consistent with the binary content-mode path (see KnativeHttpHeaderFilterStrategy). + private final HeaderFilterStrategy headerFilterStrategy = new DefaultHeaderFilterStrategy(); + protected AbstractCloudEventProcessor(CloudEvent cloudEvent) { this.cloudEvent = cloudEvent; } @@ -72,6 +80,17 @@ public Processor consumer(KnativeEndpoint endpoint, KnativeResource service) { protected abstract void decodeStructuredContent(Exchange exchange, Map content); + /** + * Maps a structured-mode CloudEvent extension field to a message header, applying the header filter strategy so + * that internal Camel ({@code Camel*}) headers are filtered consistently with the binary content-mode path. + */ + protected void mapExtensionAsHeader(Message message, Exchange exchange, String key, Object value) { + final String headerName = key.toLowerCase(Locale.US); + if (!headerFilterStrategy.applyFilterToExternalHeaders(headerName, value, exchange)) { + message.setHeader(headerName, value); + } + } + @Override public Processor producer(KnativeEndpoint endpoint, KnativeResource service) { final CloudEvent ce = cloudEvent(); diff --git a/components/camel-knative/camel-knative-component/src/main/java/org/apache/camel/component/knative/ce/CloudEventProcessors.java b/components/camel-knative/camel-knative-component/src/main/java/org/apache/camel/component/knative/ce/CloudEventProcessors.java index 7ee1e20859f7f..dcedceb295435 100644 --- a/components/camel-knative/camel-knative-component/src/main/java/org/apache/camel/component/knative/ce/CloudEventProcessors.java +++ b/components/camel-knative/camel-knative-component/src/main/java/org/apache/camel/component/knative/ce/CloudEventProcessors.java @@ -16,7 +16,6 @@ */ package org.apache.camel.component.knative.ce; -import java.util.Locale; import java.util.Map; import java.util.Objects; @@ -52,11 +51,10 @@ protected void decodeStructuredContent(Exchange exchange, Map co } // - // Map every remaining field as it is (extensions). + // Map every remaining field as it is (extensions), applying the header filter + // strategy so internal Camel headers are filtered consistently with binary mode. // - content.forEach((key, val) -> { - message.setHeader(key.toLowerCase(Locale.US), val); - }); + content.forEach((key, val) -> mapExtensionAsHeader(message, exchange, key, val)); } }), v1_0_1(new AbstractCloudEventProcessor(CloudEvents.v1_0_1) { @@ -79,11 +77,10 @@ protected void decodeStructuredContent(Exchange exchange, Map co } // - // Map every remaining field as it is (extensions). + // Map every remaining field as it is (extensions), applying the header filter + // strategy so internal Camel headers are filtered consistently with binary mode. // - content.forEach((key, val) -> { - message.setHeader(key.toLowerCase(Locale.US), val); - }); + content.forEach((key, val) -> mapExtensionAsHeader(message, exchange, key, val)); } }), v1_0_2(new AbstractCloudEventProcessor(CloudEvents.v1_0_2) { @@ -106,11 +103,10 @@ protected void decodeStructuredContent(Exchange exchange, Map co } // - // Map every remaining field as it is (extensions). + // Map every remaining field as it is (extensions), applying the header filter + // strategy so internal Camel headers are filtered consistently with binary mode. // - content.forEach((key, val) -> { - message.setHeader(key.toLowerCase(Locale.US), val); - }); + content.forEach((key, val) -> mapExtensionAsHeader(message, exchange, key, val)); } }); 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 6cdd820881d2c..9705f0b3ef970 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 @@ -379,6 +379,61 @@ void testConsumeStructuredContent(CloudEvent ce) throws Exception { mock.assertIsSatisfied(); } + @ParameterizedTest + @EnumSource(CloudEvents.class) + void testConsumeStructuredContentFiltersInternalHeaders(CloudEvent ce) throws Exception { + configureKnativeComponent( + context, + ce, + sourceEndpoint( + "myEndpoint", + Map.of( + Knative.KNATIVE_CLOUD_EVENT_TYPE, "org.apache.camel.event", + Knative.CONTENT_TYPE, "text/plain"))); + + RouteBuilder.addRoutes(context, b -> { + b.from("knative:endpoint/myEndpoint") + .to("mock:ce"); + }); + + context.start(); + + MockEndpoint mock = context.getEndpoint("mock:ce", MockEndpoint.class); + mock.expectedBodiesReceived("test"); + mock.expectedMessageCount(1); + // a regular CloudEvent extension must still be propagated as a message header + mock.expectedHeaderReceived("myextension", "myvalue"); + // an extension that resolves to an internal Camel header must be filtered out and must not + // be able to inject a framework control header (headers are matched case-insensitively) + mock.expectedMessagesMatches(e -> e.getMessage().getHeader(Exchange.FILE_NAME) == null); + + if (Objects.equals(CloudEvents.v1_0.version(), ce.version()) + || Objects.equals(CloudEvents.v1_0_1.version(), ce.version()) + || Objects.equals(CloudEvents.v1_0_2.version(), ce.version())) { + given() + .contentType(Knative.MIME_STRUCTURED_CONTENT_MODE) + .body( + Map.of( + "specversion", ce.version(), + "type", "org.apache.camel.event", + "id", "myEventID", + "source", "/somewhere", + "datacontenttype", "text/plain", + "myextension", "myvalue", + "camelfilename", "injected.txt", + "data", "test"), + ObjectMapperType.JACKSON_2) + .when() + .post() + .then() + .statusCode(200); + } else { + throw new IllegalArgumentException("Unknown CloudEvent spec: " + ce.version()); + } + + mock.assertIsSatisfied(); + } + @ParameterizedTest @EnumSource(CloudEvents.class) void testConsumeContent(CloudEvent ce) throws Exception { diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index 859c222605bc6..4b7070128c302 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -214,6 +214,17 @@ the message, consistent with the inbound header filtering already performed by t Ordinary application headers are unaffected. If a route relied on `Camel*` headers being propagated from the MIME content, set them explicitly after unmarshalling. +=== camel-knative - structured-mode CloudEvent header filtering + +When consuming a CloudEvent in structured content mode (`application/cloudevents+json`), the Knative component now +applies a `HeaderFilterStrategy` to the event fields (extensions) mapped from the payload onto the Camel message. +Camel-internal headers (the `Camel*` namespace, matched case-insensitively) present as structured-event fields are +no longer mapped onto the message, consistent with the inbound header filtering already performed on the binary +content-mode / HTTP header path. + +Ordinary CloudEvent extension attributes are unaffected. If a route relied on `Camel*`-named fields being +propagated from the structured payload, set them explicitly after consuming the event. + === camel-pinecone The `tls` endpoint option now correctly documents its default as `false` (previously the catalog