Skip to content

Bulk publish: per-message metadata in publish_events - #1228

Merged
sicoyle merged 4 commits into
dapr:mainfrom
nelson-parente:bulk-publish-entry-metadata
Sep 24, 2026
Merged

sicoyle merged 4 commits into
dapr:mainfrom
nelson-parente:bulk-publish-entry-metadata

Conversation

@nelson-parente

@nelson-parente nelson-parente commented Sep 23, 2026 •

Copy link
Copy Markdown
Contributor

Description

publish_events built each BulkPublishRequestEntry from entry_id, event and content_type only, so per-message metadata such as partitionKey or a Kafka key could not be set. The runtime also merges request-level metadata into an entry only when that entry already carries metadata, so publish_metadata never reached the cloud event envelope either. On 1.18.4, ttlInSeconds sets expiration with publish_event but not with publish_events.

This PR:

  • Adds BulkPublishEntry(event, metadata=None, content_type=None, entry_id=None), exported from dapr.clients. The shape mirrors go-sdk PublishEventsEvent and java-sdk BulkPublishEntry. Plain bytes and str entries keep working unchanged. An event of any other type raises TypeError. The class defines __eq__ and __repr__.
  • Copies publish_metadata onto every entry. Entry keys override request keys, which matches the runtime merge order. The request-level metadata is still sent because the runtime reads rawPayload from it.
  • Shares one entry builder between the sync and async clients so the two cannot drift.
  • Updates the pubsub-simple example to wrap one bulk event in BulkPublishEntry. The validated output is unchanged.
from dapr.clients import BulkPublishEntry, DaprClient

with DaprClient() as d:
    d.publish_events(
        pubsub_name='pubsub',
        topic_name='TOPIC_A',
        data=[
            BulkPublishEntry(event='{"n": 1}', metadata={'partitionKey': 'tenant-a'}),
            BulkPublishEntry(event='{"n": 2}', metadata={'partitionKey': 'tenant-b'}),
            '{"n": 3}',
        ],
        data_content_type='application/json',
        publish_metadata={'ttlInSeconds': '60'},
    )

Tests:

  • Unit tests (sync and async) assert on the BulkPublishRequest the fake sidecar recorded: entry metadata, request metadata copied onto every entry, entry-over-request precedence, caller entry IDs, content type precedence, type defaults, invalid event type.
  • The integration tests, sync and async, publish a plain entry and a BulkPublishEntry with ttlInSeconds and assert that both received messages carry the expiration extension. Redis pub/sub has no native TTL, so expiration appears only when the metadata reaches the entry. Verified locally against runtime 1.17.2: with the copy disabled, both tests fail on the plain entry.

CI note: codecov/project reports -0.07% with the patch at 100% coverage. The two clients lost their inline entry loops, which were covered lines, and the aio client's ratio shifts by 11 lines. No remaining line lost coverage.

RELEASE NOTE: ADD Per-message metadata support in publish_events bulk publish.

Issue reference

Please reference the issue this PR will close: #1214

Checklist

Please make sure you've completed the relevant tasks for this PR, out of the following list:

🤖 Generated with Claude Code

publish_events built each BulkPublishRequestEntry from entry_id, event and
content_type only, so per-message metadata such as partitionKey could not be
set. The runtime also merges request-level metadata into an entry only when
that entry already carries metadata, so publish_metadata never reached the
cloud event envelope either: ttlInSeconds set expiration on publish_event but
not on publish_events.

- Add BulkPublishEntry(event, metadata, content_type, entry_id), exported from
  dapr.clients, mirroring go-sdk PublishEventsEvent and java-sdk
  BulkPublishEntry. Plain bytes and str entries keep working.
- Copy publish_metadata onto every entry. Entry keys override request keys,
  matching the runtime merge order. The request-level metadata is still sent,
  because the runtime reads rawPayload from it.
- Share one entry builder between the sync and async clients.
- Unit tests assert on the recorded BulkPublishRequest. The integration test
  proves ttlInSeconds reaches a plain entry through the expiration extension.

Fixes dapr#1214

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Signed-off-by: Nelson Parente <nelson_parente@live.com.pt>
@codecov

codecov Bot commented Sep 23, 2026 •

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 83.89%. Comparing base (6215f4e) to head (d54d25a).

Additional details and impacted files
@@            Coverage Diff             @@
##             main    #1228      +/-   ##
==========================================
- Coverage   83.91%   83.89%   -0.03%     
==========================================
  Files         123      123              
  Lines       10260    10265       +5     
==========================================
+ Hits         8610     8612       +2     
- Misses       1650     1653       +3     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

- Port the plain-entry and type-default unit tests to the async client so
  both clients carry the same coverage of the shared entry builder.
- Document that a BulkPublishEntry fixes its entry_id at construction, so
  one instance must not appear twice in the same data sequence.
- Reuse _next_message in the existing streaming integration test instead
  of keeping two copies of the thread-and-future pattern.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Signed-off-by: Nelson Parente <nelson_parente@live.com.pt>
@nelson-parente
nelson-parente marked this pull request as ready for review September 24, 2026 09:18
@nelson-parente
nelson-parente requested review from a team as code owners September 24, 2026 09:18
@CasperGN CasperGN self-assigned this Sep 24, 2026
@CasperGN

Copy link
Copy Markdown
Contributor

A few small things, none blocking:

  1. BulkPublishEntry has no __repr__ or __eq__ (_request.py#L458-L499). A dataclass would give both, and a failed entry is easier to read in a log or a test. The entry_id default could be field(default_factory=lambda: str(uuid.uuid4())). A frozen one would need __post_init__ for the metadata copy. The other request classes in this module are plain classes, so this is style, not a must.

  2. The async integration test only checks that the message arrives (test_pubsub_async.py#L85-L105). It sends ttlInSeconds but never asserts on expiration, unlike the sync test (test_pubsub.py#L139). It also has only an entry with its own metadata. So it would still pass if the async client stopped copying publish_metadata onto plain entries, which is the bug this PR fixes.

  3. The invalid-type test is only in the sync suite (test_dapr_grpc_client.py#L427-L429). That's fine, since it tests the constructor and not the client. Just noting it, because the async suite otherwise mirrors the sync one.

  4. Docs: "Extended the documentation" is ticked, but only the docstrings and the example changed. The Python client page in dapr/docs covers only publish_event under Publish messages. The Python tab on the bulk publish page uses requests against the HTTP API, not the SDK. A short publish_events + BulkPublishEntry section there would be worth a follow-up docs PR.

@sicoyle sicoyle left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

one comment from me - thanks!!

Comment thread dapr/clients/grpc/_request.py Outdated
ValueError: event is not bytes or str.
"""
if not isinstance(event, (bytes, str)):
raise ValueError(f'invalid type for event {type(event)}')

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
raise ValueError(f'invalid type for event {type(event)}')
raise TypeError(f'invalid type for event {type(event)}')

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in d54d25a. BulkPublishEntry now raises TypeError. The docstrings and the unit tests, sync and async, follow.

nelson-parente and others added 2 commits September 24, 2026 16:36
Resolves the conflict in tests/integration/test_pubsub.py. The next_message None loop from dapr#1234 moves into the shared _next_message helper, so the bulk metadata test gets it too.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Signed-off-by: Nelson Parente <nelson_parente@live.com.pt>
- BulkPublishEntry raises TypeError for an event that is not bytes or str. It raised ValueError before.
- BulkPublishEntry defines __eq__ and __repr__. A failed entry is now readable in a log or in a test failure.
- The async integration test mirrors the sync test. It publishes a plain entry and an entry with its own metadata, and asserts the expiration extension on both. Without the publish_metadata copy, the plain entry fails.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Signed-off-by: Nelson Parente <nelson_parente@live.com.pt>
@nelson-parente

Copy link
Copy Markdown
Contributor Author

Thanks Casper. All four points, in d54d25a:

  1. __eq__ and __repr__ added. I kept the plain class to match the other request classes in _request.py. A unit test covers equality, inequality on entry_id, and the repr string.
  2. The async integration test now mirrors the sync one. It publishes a plain entry and an entry with its own metadata, and asserts expiration on both. I ran a mutation check locally with the publish_metadata copy removed: the sync test and the async test both fail on the plain entry.
  3. Agreed. The constructor test stays in the sync suite only.
  4. The docs are in Bulk publish: Python tab uses the SDK with per-entry metadata docs#5336. The Python tab on the bulk publish page already uses publish_events with BulkPublishEntry. I added a publish_events example to the Python client page in the same PR, so both pages you named are covered. The checklist item now points at that PR.

Also merged upstream main (#1234) to clear the conflict in tests/integration/test_pubsub.py. The next_message() None loop moved into the shared _next_message helper.

@sicoyle sicoyle left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM - thanks!

@sicoyle
sicoyle added this pull request to the merge queue Sep 24, 2026
Merged via the queue into dapr:main with commit fb229bc Sep 24, 2026
22 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Bulk publish: per-message metadata in publish_events

3 participants