GCP pubsub eventsource keep retrying when message body is not json

[minzetaos]

Describe the bug
I was running some test around GCP pubsub eventsource, and noticed the eventsource pod would keep retrying and create endless amount of logs if the published message body is not json. This could build up huge amount of cost and garbage the logs if someone accidentally sends a message in the wrong format.

Additionally, I couldn't stop eventsource generating these logs. I tried to ack the msg but no msg showed up on the GCP console for this subscription. I suspect it's because eventsource was keeping pulling the msg and caused nothing to show on the console end.

Below are msg I received every second:

{"level":"info","ts":1738370587.8223758,"logger":"argo-events.eventsource","caller":"gcppubsub/start.go:110","msg":"received GCP PubSub Message from t
opic","eventSourceName":"gcp-pubsub","eventSourceType":"pubsub","eventName":"xxx","topic":"xxx-workflows-topic","topicProjectID":"gcp-xxx,"projectID":"xxx","subscriptionID":"xxx-1790019940"}

{"level":"error","ts":1738370587.822445,"logger":"argo-events.eventsource","caller":"gcppubsub/start.go:123","msg":"failed to marshal the event data",
"eventSourceName":"gcp-pubsub","eventSourceType":"pubsub","eventName":"xxx","topic":"xxx-workflows-topic","topicProjectID":"xxx","projectID":"xxx","subscriptionID":"xxx-1790019940","error":"json: error calling MarshalJSON for type *json.RawMessage: invalid character 'e' in literal true (expecting 'r')","stacktrace":"github.com/argoproj/argo-events/pkg/eventsources/sources/gcppubsub.(*EventListener).StartListening.func1\n\t/home/runner/work/argo-events/argo-events/pkg/eventsources/sources/gcppubsub/start.go:123\ncloud.google.com/go/pubsub.(*Subscription).Receive.func2.1\n\t/home/runner/go/pkg/mod/cloud.google.com/go/[email protected]/subscription.go:1403\ncloud.google.com/go/pubsub/internal/scheduler.(*ReceiveScheduler).Add.func1\n\t/home/runner/go/pkg/mod/cloud.google.com/go/[email protected]/internal/scheduler/receive_scheduler.go:84"}

To Reproduce
Steps to reproduce the behavior:
I used the default eventbus and the eventsource was set up like this:

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: gcp-pubsub
  namespace: argo-events
spec:
  template:
    serviceAccountName: xxx-argo-events
  pubSub:
    dev-dev-sub:
      jsonBody: true
      projectID: xxx
      topic: xxx

Then I tried to publish a text message with the msg body test from the GCP pubsub console.

When the msg was published, the eventsource went crazy.

Expected behavior
If the published msg format is wrong, it should stop messaging after several tries.

Environment:

• Kubernetes: [v1.30.8]
• Argo: [v3.6.2]
• Argo Events: [v1.9.5]

---

Message from the maintainers:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.