Topic multi-partition writer (write-by-key)#864
Draft
vgvoleg wants to merge 1 commit into
Draft
Conversation
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.
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.
Comment on lines
+139
to
+146
| def add_partitions(self, partitions: List[PartitionInfo]) -> None: | ||
| for i, p in enumerate(partitions): | ||
| from_bound = p.key_range.from_bound if p.key_range is not None else b"" | ||
| if i > 0 and not from_bound: | ||
| raise ValueError("non-first partition without a from_bound key range") | ||
| self._partitions.append((from_bound, p.partition_id)) | ||
| self._partitions.sort(key=lambda x: x[0]) | ||
|
|
Comment on lines
+159
to
+165
| # First partition whose from_bound is strictly greater than the hashed key; | ||
| # the owning partition is the previous one. | ||
| bounds = [b for b, _ in self._partitions] | ||
| idx = bisect.bisect_right(bounds, hashed) | ||
| if idx == 0: | ||
| raise RuntimeError("inconsistent partition bounds: lower-bound search returned 0") | ||
| return self._partitions[idx - 1][1] |
Comment on lines
+108
to
+115
| def add_partitions(self, partitions: List[PartitionInfo]) -> None: | ||
| for p in partitions: | ||
| kr = p.key_range | ||
| if kr is not None and (kr.from_bound or kr.to_bound): | ||
| raise ValueError("PublicPartitionByKeyKafka does not support partition key ranges") | ||
| self._partitions.append(p.partition_id) | ||
| self._partitions.sort() | ||
|
|
Comment on lines
+315
to
+319
| When an auto-partitioned partition is split (one into two) or merged (two into one), the | ||
| multi-writer re-describes the topic, routes subsequent keys to the new child partition(s), and | ||
| transparently resends the messages that were still in flight to the retired partition(s). | ||
| Messages already persisted before the change are not resent (so no duplicates), and per-key | ||
| ordering is preserved. |
| @@ -1,3 +1,5 @@ | |||
| * Add a topic multi-partition writer (`topic_client.multiwriter(...)`) that routes messages across partitions by their `key`, with Kafka-hash and key-range partition choosers, and transparently resends in-flight messages to child partitions on an auto-partition split (no loss, no duplicates); expose partition `key_range` on `describe_topic` results | |||
| partition_chooser: Optional[TopicWriterPartitionChooser] = None, # default - route by key (Kafka hash) | ||
| auto_seqno: bool = True, | ||
| auto_created_at: bool = True, | ||
| codec: Optional[TopicCodec] = None, # default mean auto-select |
| partition_chooser: Optional[TopicWriterPartitionChooser] = None, # default - route by key (Kafka hash) | ||
| auto_seqno: bool = True, | ||
| auto_created_at: bool = True, | ||
| codec: Optional[TopicCodec] = None, # default mean auto-select |
Comment on lines
+462
to
+466
| async with self._lock: | ||
| assert self._chooser is not None # resolved by _init(), awaited above | ||
| for message in converted: | ||
| partition_id = self._chooser.choose_partition(message) | ||
| writer = await self._get_or_create_writer(partition_id) |
Comment on lines
+364
to
+368
| async with self._lock: | ||
| if partition_id not in self._partitions: | ||
| return # already handled by a sibling parent's event | ||
| assert self._chooser is not None # resolved by _init() before any repartition | ||
|
|
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.