feat(pubsub/pulsar): register CloudEvents envelope Avro schema with Pulsar Schema Registry#4302
Conversation
javier-aliaga
changed the title
There was a problem hiding this comment.
Pull request overview
The PR implements registration of CloudEvents envelope Avro schemas with Pulsar Schema Registry. Previously, when Dapr published to Pulsar with CloudEvents wrapping enabled (the default), the schema registered with Pulsar was only the inner domain event schema, creating a mismatch with the actual wire format. This fix wraps the inner schema in a CloudEvents envelope Avro schema before registration.
Changes:
- New CloudEvents schema wrapping functionality in
cloudevents_schema.gothat embeds inner Avro schemas as thedatafield of a CloudEvents envelope - Enhanced
schemaMetadatastructure to store both the inner schema codec and CloudEvents envelope codec - Updated producer and consumer registration to use the CloudEvents envelope schema when available
- Added explicit error handling to reject
rawPayload=trueon Avro schema topics using CloudEvents envelopes - Comprehensive test coverage for the new functionality
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
pubsub/pulsar/cloudevents_schema.go |
New file implementing CloudEvents Avro schema wrapping logic |
pubsub/pulsar/cloudevents_schema_test.go |
New comprehensive test file for schema wrapping with edge cases |
pubsub/pulsar/metadata.go |
Enhanced schemaMetadata struct with CE envelope fields |
pubsub/pulsar/pulsar.go |
Updated schema initialization, producer/consumer registration, and publish validation |
pubsub/pulsar/pulsar_test.go |
Added test helpers and new test cases for CE envelope handling |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
c3b6b2e to
28f9a05
Compare
1f77e2a to
67e42ea
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
1c43d13 to
ad96bbe
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
6e2accb to
f417293
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Document the new behavior where Dapr wraps user-provided Avro schemas inside a CloudEvents envelope before registering with the Pulsar Schema Registry. Add the new `<topic-name>.rawPayload` metadata field for topics that need to skip CE wrapping and register the inner schema directly. Ref: dapr/components-contrib#4302 Signed-off-by: Javier Aliaga <[email protected]>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 9 out of 9 changed files in this pull request and generated no new comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 9 out of 9 changed files in this pull request and generated 2 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 9 out of 9 changed files in this pull request and generated no new comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
…ulsar Schema Registry When Dapr publishes to Pulsar with CloudEvents wrapping (the default), the actual wire format is a CloudEvents envelope containing the user's domain event in the `data` field. Previously, the schema registered with the Pulsar Schema Registry was the inner domain event schema, causing a mismatch that breaks external consumers (e.g., Apache Flink) and may cause Pulsar to reject messages when schema enforcement is enabled. This change wraps the user-provided Avro schema inside a CloudEvents envelope Avro schema before registering it with the broker. The envelope includes all standard CE attributes (id, source, specversion, type), optional attributes (datacontenttype, subject, time), Dapr extensions (topic, pubsubname, traceid, traceparent, tracestate, expiration), and the data/data_base64 fields. The inner domain schema is embedded as the type of the nullable `data` union field. Using rawPayload=true on a topic with an Avro schema now returns an explicit error, since the broker enforces a single schema per topic and mixing CE-wrapped and raw payloads is undefined behavior. Signed-off-by: Javier Aliaga <[email protected]>
3fed7d5 to
71d89c3
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 9 out of 9 changed files in this pull request and generated no new comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
The backport to To backport manually, run these commands in your terminal: # Fetch latest updates from GitHub
git fetch
# Create a new working tree
git worktree add .worktrees/backport-release-1.16 release-1.16
# Navigate to the new working tree
cd .worktrees/backport-release-1.16
# Create a new branch
git switch --create backport-4302-to-release-1.16
# Cherry-pick the merged commit of this pull request and resolve the conflicts
git cherry-pick -x --mainline 1 4ad2c5bbe7873b9bb6044c2f1e736b47b55a145e
# Push it to GitHub
git push --set-upstream origin backport-4302-to-release-1.16
# Go back to the original working tree
cd ../..
# Delete the working tree
git worktree remove .worktrees/backport-release-1.16Then, create a pull request where the |
|
The backport to To backport manually, run these commands in your terminal: # Fetch latest updates from GitHub
git fetch
# Create a new working tree
git worktree add .worktrees/backport-release-1.17 release-1.17
# Navigate to the new working tree
cd .worktrees/backport-release-1.17
# Create a new branch
git switch --create backport-4302-to-release-1.17
# Cherry-pick the merged commit of this pull request and resolve the conflicts
git cherry-pick -x --mainline 1 4ad2c5bbe7873b9bb6044c2f1e736b47b55a145e
# Push it to GitHub
git push --set-upstream origin backport-4302-to-release-1.17
# Go back to the original working tree
cd ../..
# Delete the working tree
git worktree remove .worktrees/backport-release-1.17Then, create a pull request where the |
…ulsar Schema Registry (dapr#4302) Signed-off-by: Javier Aliaga <[email protected]>
…ulsar Schema Registry (dapr#4302)
…ulsar Schema Registry (dapr#4302) Signed-off-by: Javier Aliaga <[email protected]>
…ulsar Schema Registry (dapr#4302) Signed-off-by: Javier Aliaga <[email protected]>
Document the new behavior where Dapr wraps user-provided Avro schemas inside a CloudEvents envelope before registering with the Pulsar Schema Registry. Add the new `<topic-name>.rawPayload` metadata field for topics that need to skip CE wrapping and register the inner schema directly. Ref: dapr/components-contrib#4302 Signed-off-by: Javier Aliaga <[email protected]>
Document the new behavior where Dapr wraps user-provided Avro schemas inside a CloudEvents envelope before registering with the Pulsar Schema Registry. Add the new `<topic-name>.rawPayload` metadata field for topics that need to skip CE wrapping and register the inner schema directly. Ref: dapr/components-contrib#4302 Signed-off-by: Javier Aliaga <[email protected]>
Document the new behavior where Dapr wraps user-provided Avro schemas inside a CloudEvents envelope before registering with the Pulsar Schema Registry. Add the new `<topic-name>.rawPayload` metadata field for topics that need to skip CE wrapping and register the inner schema directly. Ref: dapr/components-contrib#4302 Signed-off-by: Javier Aliaga <[email protected]>
Document the new behavior where Dapr wraps user-provided Avro schemas inside a CloudEvents envelope before registering with the Pulsar Schema Registry. Add the new `<topic-name>.rawPayload` metadata field for topics that need to skip CE wrapping and register the inner schema directly. Ref: dapr/components-contrib#4302 Signed-off-by: Javier Aliaga <[email protected]>
Description
When Dapr publishes to Pulsar with CloudEvents wrapping enabled (the default), the wire format is a CloudEvents envelope containing the user's domain event in the
datafield. Previously, the Avro schemaregistered with the Pulsar Schema Registry was the inner domain event schema, not the CloudEvents envelope — causing a mismatch between the schema in the registry and the actual messages stored in the topic.
This PR fixes the mismatch by wrapping the user-provided Avro schema inside a CloudEvents envelope Avro schema before registering it with the broker.
rawschema=trueon publish request.avroschema.avroschemawithout.rawschemaflag.avroschemawith.rawschema=trueflagChanges
cloudevents_schema.go): newwrapInCloudEventsAvroSchema()embeds the user's inner schema as thedatafield of a CloudEvents envelope record, following theCloudEvents Avro format spec
metadata.go):schemaMetadatanow stores both the inner codec (codec/value) and the CE envelope codec (ceCodec/ceValue)pulsar.go): use the CE envelope schema when available for bothPublishandSubscribepulsar.go): select the correct codec (CE envelope vs inner) at publish time; rejectrawschema=trueon CE-wrapped topics with a clear errorrawschemaflag (pulsar.go,metadata.yaml): new<topic-name>.rawSchema=truemetadata option skips CE envelope wrapping for dedicated raw-payload topics, registering the inner schemadirectly with the broker
Topic-level
rawschemaconfigurationPulsar enforces a single schema per topic. Since the CE envelope schema differs from the inner schema,
rawsc hema=truemessages cannot be sent to a CE-wrapped topic. This PR provides two options:Default (CE-wrapped topic):
Publishing with rawschema=true to a CE-wrapped topic returns:
rawschema=true is not compatible with Avro schema topics using CloudEvents envelope; use a separate topic for raw payloads
Issue reference
We strive to have all PR being opened based on an issue, where the problem or feature have been discussed prior to implementation.
Please reference the issue this PR will close: #[issue number]
Checklist
Please make sure you've completed the relevant tasks for this PR, out of the following list:
Note: We expect contributors to open a corresponding documentation PR in the dapr/docs repository. As the implementer, you are the best person to document your work! Implementation PRs will not be merged until the documentation PR is opened and ready for review.