fix(pubsub/pulsar): decode Avro binary payload on subscribe path

[nelson-parente]

Summary

PR #4244 added Avro schema validation on the publish path but did not update handleMessage on the subscribe side.

The Pulsar Go client returns raw Avro binary bytes from msg.Payload() even when a schema consumer is configured. The Dapr runtime then tries json.Unmarshal on those bytes and fails:

error deserializing cloud event in pubsub <component> and topic <topic>:
invalid character '\x06' looking for beginning of value
scope=dapr.runtime.processor.subscription

\x06 is the first byte of the Avro binary-encoded payload (string length 3 for specversion = "1.0" encoded as zigzag varint). Every message on an Avro-enforced topic gets stuck in a permanent retry loop and is never delivered to the application.

Root Cause

handleMessage always passes msg.Payload() (raw bytes) directly:

pubsubMsg := pubsub.NewMessage{
    Data: msg.Payload(), // always raw bytes, never decoded
    ...
}

The goavro.Codec required for decoding is already compiled and cached in p.metadata.internalTopicSchemas[topic].codec (added by #4244). It is simply never used on the receive path.

Fix

When an Avro schema is registered for the incoming topic, decode the binary payload to native Go types using the cached codec, then re-encode as JSON before passing to the handler. This mirrors the publish path:

Direction	Operation
Publish (#4244)	NativeFromTextual (JSON → native) → msg.Value (Avro binary via SDK)
Subscribe (this PR)	NativeFromBinary (Avro binary → native) → TextualFromNative (→ JSON)

No new dependencies. The fix is ~15 lines in handleMessage only.

Tests

Adds three unit tests for the subscribe decode path:

• Round-trip: Avro binary → handleMessage → JSON delivered to handler ✅
• Decode error: truncated payload → Nack + error returned, handler not called ✅
• No schema: raw payload passed through unchanged (existing behaviour preserved) ✅

Affected Files

• pubsub/pulsar/pulsar.go — handleMessage only
• pubsub/pulsar/pulsar_test.go — new tests

Fixes the subscribe-side gap introduced by #4244.

[MyMirelHub]

Can we also add a certification test with pulsar avro - see one for equivalent json schema here https://github.com/dapr/components-contrib/blob/main/tests/certification/pubsub/pulsar/components/auth-none/consumer_four/pulsar.yaml

[Copilot]

Pull request overview

This PR fixes Pulsar pubsub subscriptions for topics configured with Avro schemas by decoding Avro-binary payloads back into JSON in the subscribe (handleMessage) path, aligning behavior with the Avro validation/encoding added on the publish path in #4244.

Changes:

• Decode Avro binary payloads to JSON in pubsub/pulsar subscribe handling when an Avro schema is configured for the topic.
• Add unit tests covering Avro decode success/failure and passthrough behavior when no (or non-Avro) schema is present.
• Extend Pulsar certification tests + component manifests to validate Avro schema publish/subscribe round-trip.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
pubsub/pulsar/pulsar.go	Decode Avro-binary payloads to JSON before handing messages to the runtime when topic schema protocol is Avro.
pubsub/pulsar/pulsar_test.go	Add unit tests for Avro decode behavior in handleMessage plus ack/nack expectations.
tests/certification/pubsub/pulsar/pulsar_test.go	Add a certification scenario for Avro schema topics and refactor schema-message publishing helpers.
tests/certification/pubsub/pulsar/components/auth-oauth2/consumer_nine/pulsar.yml.tmpl	Add .avroschema metadata for the new certification consumer configuration (OAuth2).
tests/certification/pubsub/pulsar/components/auth-none/consumer_nine/pulsar.yaml	Add .avroschema metadata for the new certification consumer configuration (no auth).
tests/certification/pubsub/pulsar/README.md	Document the added Avro schema encode/decode round-trip certification coverage.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

[Copilot]

Pull request overview

This PR fixes Pulsar pubsub subscriptions for topics configured with .avroschema by decoding Avro binary payloads on the subscribe path before handing data to the Dapr runtime, preventing JSON deserialization failures and retry loops.

Changes:

• Decode Avro binary payloads in handleMessage using the cached goavro.Codec, then convert to JSON bytes before invoking the handler.
• Add unit tests covering the Avro decode path (success, decode failure, passthrough for non-Avro/no-schema topics, metadata preservation).
• Extend Pulsar certification tests/components to validate Avro encode/decode round-trip behavior end-to-end.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
pubsub/pulsar/pulsar.go	Decode Avro binary payloads to JSON in handleMessage when an Avro schema is configured for the topic.
pubsub/pulsar/pulsar_test.go	Adds unit tests for subscribe-side Avro decoding and related behavior (nack on decode error, passthrough, metadata).
tests/certification/pubsub/pulsar/pulsar_test.go	Adds a certification test and helpers to validate Avro schema round-trip on publish/subscribe.
tests/certification/pubsub/pulsar/components/auth-oauth2/consumer_nine/pulsar.yml.tmpl	Adds a new OAuth2 component variant configured with .avroschema metadata for certification.
tests/certification/pubsub/pulsar/components/auth-none/consumer_nine/pulsar.yaml	Adds a new no-auth component variant configured with .avroschema metadata for certification.
tests/certification/pubsub/pulsar/README.md	Documents the new Avro schema encode/decode certification scenario.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.