CAMEL-24084: Apply header filter strategy to structured-mode CloudEvent extension headers (#24724)

The Knative consumer filters inbound HTTP headers through KnativeHttpHeaderFilterStrategy,
but structured content mode (application/cloudevents+json) mapped the remaining event fields
(extensions) onto the message headers without a HeaderFilterStrategy, inconsistent with the
binary/HTTP-header path. Route structured-mode extension fields through a HeaderFilterStrategy
so Camel* headers (matched case-insensitively) are filtered consistently on both content modes.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: oscerd

components/camel-knative/camel-knative-component/src/main/java/org/apache/camel/component/knative/ce/AbstractCloudEventProcessor.java

@@ -19,23 +19,31 @@
 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;
 
 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<String, Object> 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();

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<String, Object> 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<String, Object> 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<String, Object> 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));
         }
     });
 

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 {

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc

@@ -227,6 +227,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