fix: Include CE attributes in sink headers

- Forward CloudEvent attributes to the sink
- Prevent default Camel header filter strategy removing the CE headers

Author: christophd

aws-eventbridge-sink/src/main/resources/camel/sink.yaml

@@ -27,6 +27,10 @@ spec:
       name: default
     properties:
       type: ""
+    dataTypes:
+      out:
+        scheme: http
+        format: application/cloudevents
   sink:
     ref:
       kind: Kamelet

aws-eventbridge-sink/src/test/java/dev/knative/eventing/connector/AwsEventBridgeSinkTest.java

@@ -84,7 +84,7 @@ public void shouldConsumeEvents() {
                     .body(eventData)
                     .header("ce-id", "citrus:randomPattern([0-9A-Z]{15}-[0-9]{16})@")
                     .header("ce-type", "dev.knative.eventing.aws.eventbridge")
-                    .header("ce-source", "dev.knative.eventing.aws-eventbridge-source")
+                    .header("ce-source", "aws.s3")
                     .header("ce-subject", "aws-eventbridge-source")
         );
 

aws-s3-sink/src/main/resources/camel/sink.yaml

@@ -27,6 +27,10 @@ spec:
       name: default
     properties:
       type: ""
+    dataTypes:
+      out:
+        scheme: http
+        format: application/cloudevents
   sink:
     ref:
       kind: Kamelet

aws-sns-sink/src/main/resources/camel/sink.yaml

@@ -27,6 +27,10 @@ spec:
       name: default
     properties:
       type: ""
+    dataTypes:
+      out:
+        scheme: http
+        format: application/cloudevents
   sink:
     ref:
       kind: Kamelet

aws-sns-sink/src/test/java/dev/knative/eventing/connector/AwsSnsSinkTest.java

@@ -81,6 +81,7 @@ public void shouldConsumeEvents() {
                     .header("ce-type", "dev.knative.eventing.aws.sns")
                     .header("ce-source", "dev.knative.eventing.aws-sns-source")
                     .header("ce-subject", "aws-sns-source")
+                    .header("ce-apiVersion", "v1")
         );
 
         tc.when(
@@ -110,7 +111,14 @@ private void verifySnsData(TestContext context) {
 
         try {
             Assertions.assertEquals(1, receiveMessageResponse.messages().size());
-            Assertions.assertTrue(receiveMessageResponse.messages().get(0).body().contains(snsData));
+            String body = receiveMessageResponse.messages().getFirst().body();
+            Assertions.assertTrue(body.contains("\"Message\": \"%s\"".formatted(snsData)));
+            Assertions.assertTrue(body.contains("\"Subject\": \"aws-sns-source\""));
+            Assertions.assertTrue(body.contains("ce-id"));
+            Assertions.assertTrue(body.contains("ce-type"));
+            Assertions.assertTrue(body.contains("ce-source"));
+            Assertions.assertTrue(body.contains("ce-subject"));
+            Assertions.assertTrue(body.contains("ce-apiVersion"));
         } catch (AssertionFailedError error) {
             throw new CitrusRuntimeException("SNS data verification failed", error);
         }

aws-sqs-sink/src/main/resources/camel/sink.yaml

@@ -27,6 +27,10 @@ spec:
       name: default
     properties:
       type: ""
+    dataTypes:
+      out:
+        scheme: http
+        format: application/cloudevents
   sink:
     ref:
       kind: Kamelet

log-sink/src/main/resources/camel/sink.yaml

@@ -27,6 +27,10 @@ spec:
       name: default
     properties:
       type: ""
+    dataTypes:
+      out:
+        scheme: http
+        format: application/cloudevents
   sink:
     ref:
       kind: Kamelet