IntegrationSink SNS converts and publishes messages in a non-standard format

[darkgnotic]

Describe the bug

(I am not 100% where the bug is happening, so apologies if this is a configuration error on my part)

When sending events through the IntegrationSink SNS option, the topic's Message field appears to be formatted in a non-JSON object format (e.g. with '=' as the key/value delimiter):

{
  "Type": "Notification",
  "MessageId": "70f6a303-3572-56c0-adba-c25935d9a107",
  "TopicArn": "arn:aws:sns:us-east-1:...",
  "Subject": "/apis/v1/namespaces/foo/pods/bar",
  "Message": "{apiVersion=v1, kind=Pod, metadata={creationTimestamp=2025-10-03T02:37:45Z, generateName=...",
  ...
}

The IntegrationSink Log option, however, correctly outputs JSON bodies:

Body: {"apiVersion":"v1","kind":"Pod","metadata":{"creationTimestamp":"2025-10-03T02:37:45Z","generateName":"..."}

I am also able to reproduce this with events from my application, so it is not limited to events from the ApiServerSource:

SNS sink:

{
  "Type": "Notification",
  "MessageId": "e770e27c-04b1-5345-8a0b-83253ed340c8",
  "TopicArn": "arn:aws:sns:us-east-1:...",
  "Message": "{type=zero/events/status/replication/v1, component=replication, status=OK, stage=Indexing, description=Creating 22 indexes, time=2025-10-03T02:42:37.592Z, ..."

Log sink:

Body: {"type":"zero/events/status/replication/v1","component":"replication","status":"OK","stage":"Indexing","description":"Creating 22 indexes","time":"2025-10-03T02:42:37.592Z", ...}

--

Expected behavior

The Message that the SNS option publishes to the Topic should be the same data and format that it received.

To Reproduce

I am able to reproduce this with the built in ApiServerSource as well as with custom CloudEvents.

Knative release version

I'm using knative-eventing v1.19.0

Additional context

It's possible that it is related to the Kamlet setup, similar to (and perhaps even fixed by?) knative-extensions/eventing-integrations#103.

This is just a guess though ... I have not studied the code well enough to make a better diagnosis.

Your help is appreciated! 🙏

[darkgnotic]

One thing to add is that whatever this format is, I don't believe there is a way to parse it correctly, as it appears to allow values to contain spaces and/or the = sign, without any sort of escaping.

For instance, in the kubectl.kubernetes.io/last-applied-configuration field below, the value is a json string that has the = sign in it:

{apiVersion=apps/v1, kind=Deployment, metadata={annotations={deployment.kubernetes.io/revision=1, kubectl.kubernetes.io/last-applied-configuration={"apiVersion":"apps/v1","kind":"Deployment","metadata":{"..."},"spec": ...{"name":"OTEL_RESOURCE_ATTRIBUTES","value":"service.namespace=foo,cloud.provider=aws,cloud.account.id=17345742970..."volumes":[{"emptyDir":{"sizeLimit":"10Gi"},"name":"data"}]}}}}\n}, creationTimestamp=2025-10-03T02:37:44Z, generation=1, ...

so I haven't been able to come up with a workaround to process the data as it is currently sent.