Topic multi-partition writer (write-by-key) - #864
Open
vgvoleg wants to merge 7 commits into
Open
Conversation
There was a problem hiding this comment.
Pull request overview
Adds a new “multi-partition” topic writer API that routes messages across partitions by message key, while reusing the existing per-partition writer implementation (buffering/encoding/reconnect) and adding split/merge-aware resend logic on top. It also surfaces partition key ranges in topic descriptions to support server-accurate routing on auto-partitioned topics.
Changes:
- Add
topic_client.multiwriter(...)(sync + async), plus partition-chooser implementations (Kafka-hash and server key-range routing) and multi-writer orchestration (split/merge discovery, resend, idle sub-writer eviction). - Extend topic
DescribeTopicpublic types to include per-partitionkey_range. - Add unit + integration tests, docs, an example, and a changelog entry for the new feature.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| ydb/topic.py | Exposes multiwriter(...) API and re-exports new public types/chooser classes. |
| ydb/_topic_writer/topic_writer.py | Adds message key, split-stopping hook in settings, and internal split error type. |
| ydb/_topic_writer/topic_writer_test.py | Adds golden-vector and chooser behavior tests (murmur, Kafka/bound routing). |
| ydb/_topic_writer/topic_writer_partition_chooser.py | Implements key-based partition choosers and hashing utilities. |
| ydb/_topic_writer/topic_writer_multi_sync.py | Adds sync facade over the async multi-writer. |
| ydb/_topic_writer/topic_writer_multi_asyncio.py | Implements multi-writer routing, per-partition sub-writers, and split/merge resend logic. |
| ydb/_topic_writer/topic_writer_asyncio.py | Adds an internal hook to stop a sub-writer early on split-signaling errors. |
| ydb/_topic_writer/topic_writer_asyncio_test.py | Adds deterministic unit tests for routing, split/merge migration, eviction, and edge cases. |
| ydb/_grpc/grpcwrapper/ydb_topic.py | Maps proto PartitionKeyRange into wrapper/public types for describe results. |
| ydb/_grpc/grpcwrapper/ydb_topic_public_types.py | Adds PartitionKeyRange + PartitionInfo.key_range to the public describe model. |
| tests/topics/test_topic_writer.py | Adds integration tests for key-range exposure and write-by-key behavior (incl. split when it occurs). |
| examples/topic/multiwriter_example.py | Adds a runnable example demonstrating sync/async write-by-key usage. |
| docs/topic.rst | Documents the new write-by-key multiwriter API and chooser semantics. |
| CHANGELOG.md | Adds a user-facing entry for the new multiwriter + key_range exposure. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Routes messages across partitions by key (Kafka-hash or server key-range chooser) and transparently resends in-flight messages to child partitions on an auto-partition split, without loss or duplicates. Exposes partition key_range on describe_topic.
- take the dedup cut from the server rather than the highest ack we saw: an unpinned probe session reads last_seqno across the partition lineage, which closes the duplicate window when an ack is lost during a split - use one writer-wide seqno, so a resend keeps the message's number - give in-flight messages a terminal outcome when repartition and recovery both fail, instead of leaving their futures unresolved - own repartition tasks: coalesce per partition, cancel and await on close - fail a migration that would overwrite a colliding seqno - validate bound chooser ranges after sorting, not in argument order - raise a validation error, not TopicWriterStopped, for a missing manual seqno - lower the unclosed TopicClient message to debug
vgvoleg
force-pushed
the
topic-multiwriter
branch
from
September 2, 2026 13:35
34b82ab to
3603081
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #864 +/- ##
==========================================
+ Coverage 82.36% 83.53% +1.16%
==========================================
Files 99 102 +3
Lines 12753 13475 +722
Branches 1242 1349 +107
==========================================
+ Hits 10504 11256 +752
+ Misses 1794 1764 -30
Partials 455 455
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
- new `sync-topic-multiwriter` workload: same reader and delivery accounting, but writes by key through the multi-partition writer, so chaos runs cover key routing and the per-partition sub-writer pool - one payload stream per key rather than per partition, since a key is what the writer keeps ordered - scope topic paths by workload as well as ref: two topic workloads sharing a topic would mix into each other's ordering accounting - compute the SLO matrix from the changed files so the label starts only the affected scenarios; an unrelated diff still runs everything
- tests for the sync facade, the client factories, the key_range wrapper and the reconnector error hook, plus the orchestrator's lifecycle and error paths: 100% of the changed lines are now exercised - bound chooser: search in place instead of rebuilding a bounds list per message - Kafka chooser: reject any key range, including the fully open one a single auto-partitioned partition reports before its first split - drop an unreachable guard in the migration fail-tail loop - document that the dedup cut comes from the server, and fix a comment typo
Codecov counts partially taken branches, not just unexecuted lines, so the patch was short of 100% even with every line covered. Adds the missing sides: the guards against completing a caller's future twice, the synchronous-driver describe path, idle eviction switched off, and the coverage check rejecting children that overlap instead of extending the parent's range. Also maps SLO harness paths to the workloads they actually serve, so touching the topic jobs no longer starts the table and query scenarios.
The metric is zero almost everywhere and spikes only around a chaos fault, so comparing two such runs divides one accidental spike by another: across two runs it flagged sync-table (50%, sums 12 vs 14) and then async-topic (450%, sums 75 vs 12), each from 2 non-zero samples out of 120 with a concordance of 0.008. `direction: neutral` disables exactly that comparison while the absolute check still runs, so "retries must stay at zero" is kept. The absolute bound is restated because an exact-name entry replaces the action's `*_attempts` default outright rather than merging with it.
Walking up to the ancestors was copied from the reference implementations, but it is unnecessary here and unsafe once merges exist. Unnecessary because every move is already gated by a cut of at least that producer's server seqno, so a message that travelled has a number above it and a retired producer never grows. Unsafe because a merge child has two parents whose branches numbered independently: the maximum over them pulls in the sibling's history, which says nothing about messages that came down this branch and can be arbitrarily higher, marking unsent messages as written. Neither reference hits this — both read only the first parent and assume a single one. Removing the walk also removes the parent map it needed, and with it the case where a stable producer id prefix let a previous run's ancestor inflate the cut. Adds the architecture document the change is described in, in English and Russian.
🌋 SLO Test Results🔴 6 workload(s) tested — 1 workload(s) exceeded failure thresholds
Threshold violations: async-topic:
sync-table:
Generated by ydb-slo-action |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds
topic_client.multiwriter(...): one logical writer that routes messages across partitions by theirkey, reusing a per-partition writer under the hood.TopicWriterPartitionByKeyKafka(murmur2, Kafka-compatible) andTopicWriterPartitionByKeyBound(server key-range); adaptive default per topic.key_rangeondescribe_topic.Split path validated live against a cloud cluster (exactly-once across real splits); merge is not yet server-supported.