From 697e927341285a499056839ed593c23b8725b572 Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Mon, 5 Oct 2026 17:03:26 +0300 Subject: [PATCH 1/7] Add executable documentation examples and dedicated CI --- .github/workflows/ydb-tech.yml | 33 ++++ ydb_tech/README.md | 26 +++ ydb_tech/run.sh | 26 +++ ydb_tech/topic/README.md | 17 ++ ydb_tech/topic/async_example.py | 316 ++++++++++++++++++++++++++++++++ ydb_tech/topic/run.sh | 5 + ydb_tech/topic/sync_example.py | 303 ++++++++++++++++++++++++++++++ 7 files changed, 726 insertions(+) create mode 100644 .github/workflows/ydb-tech.yml create mode 100644 ydb_tech/README.md create mode 100755 ydb_tech/run.sh create mode 100644 ydb_tech/topic/README.md create mode 100644 ydb_tech/topic/async_example.py create mode 100755 ydb_tech/topic/run.sh create mode 100644 ydb_tech/topic/sync_example.py diff --git a/.github/workflows/ydb-tech.yml b/.github/workflows/ydb-tech.yml new file mode 100644 index 00000000..f8b2fcba --- /dev/null +++ b/.github/workflows/ydb-tech.yml @@ -0,0 +1,33 @@ +name: ydb.tech examples + +on: + pull_request: + push: + branches: [main, docs/topic-snippets] + +concurrency: + group: ydb-tech-examples-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + examples: + name: Examples (ydb_tech) + runs-on: ubuntu-latest + timeout-minutes: 15 + services: + ydb: + image: ydbplatform/local-ydb:trunk + ports: ["2136:2136", "8765:8765"] + env: + YDB_USE_IN_MEMORY_PDISKS: true + options: '-h localhost' + steps: + - uses: actions/checkout@v5 + - uses: actions/setup-python@v5 + with: + python-version: '3.10' + - run: python -m pip install -e . + - run: bash ydb_tech/run.sh diff --git a/ydb_tech/README.md b/ydb_tech/README.md new file mode 100644 index 00000000..62e80e87 --- /dev/null +++ b/ydb_tech/README.md @@ -0,0 +1,26 @@ +# Executable documentation examples + +`ydb_tech` contains the executable sources used by ydb.tech documentation. +The `Examples (ydb_tech)` CI job builds and runs every registered example against +this checkout and a local YDB instance. + +Run all examples from the repository root: + +```sh +bash ydb_tech/run.sh +``` + +Each example has its own directory and a `run.sh` entry point. The common runner +finds these entry points automatically and fails if any example fails or times out. +It requires Python 3 for portable process timeouts. Set `YDB_TECH_TIMEOUT_SECONDS` +to change the default 180-second limit per example. + +To add an example: + +1. Add a directory with executable source and a `run.sh` entry point. +2. Register its project in the SDK build if the language requires it. +3. Mark documentation regions with `[BEGIN name]` and `[END name]` comments. +4. Keep each region nonempty, unique within its file, and free of nested regions. + +No workflow changes are needed for a new example. The topic example is the first +scenario; its README describes SDK-specific build and connection settings. diff --git a/ydb_tech/run.sh b/ydb_tech/run.sh new file mode 100755 index 00000000..1b4bdcbb --- /dev/null +++ b/ydb_tech/run.sh @@ -0,0 +1,26 @@ +#!/usr/bin/env bash +set -euo pipefail +examples_root="$(cd "$(dirname "$0")" && pwd)" +repo_root="$(dirname "$examples_root")" +cd "$repo_root" +example_count=0 +for entry in "$examples_root"/*/run.sh; do + [[ -f "$entry" ]] || continue + example_count=$((example_count + 1)) + printf 'Running %s\n' "${entry%/run.sh}" + python3 - "$entry" "${YDB_TECH_TIMEOUT_SECONDS:-180}" <<'PYTHON' +import subprocess +import sys + +try: + result = subprocess.run(["bash", sys.argv[1]], timeout=int(sys.argv[2]), check=False) +except subprocess.TimeoutExpired: + print("Example exceeded its time limit: " + sys.argv[1], file=sys.stderr) + raise SystemExit(124) +raise SystemExit(result.returncode) +PYTHON +done +if [[ "$example_count" == 0 ]]; then + printf 'No executable examples found in %s\n' "$examples_root" >&2 + exit 1 +fi diff --git a/ydb_tech/topic/README.md b/ydb_tech/topic/README.md new file mode 100644 index 00000000..9d80b064 --- /dev/null +++ b/ydb_tech/topic/README.md @@ -0,0 +1,17 @@ +# Topic examples for ydb.tech + +These applications supply the synchronous and asynchronous Python snippets in the +topic reference on ydb.tech. They use the SDK from this checkout. + +From the repository root, with a local YDB instance: + +```sh +python -m pip install -e . +python ydb_tech/topic/sync_example.py +python ydb_tech/topic/async_example.py +``` + +`YDB_ENDPOINT` defaults to `grpc://localhost:2136` and `YDB_DATABASE` to `/local`. +The applications create unique topics, verify payloads and metadata, exercise +commit variants and transactions, and remove their topics afterwards. +Read and write waits are bounded; errors cause a nonzero exit. diff --git a/ydb_tech/topic/async_example.py b/ydb_tech/topic/async_example.py new file mode 100644 index 00000000..d0a2c655 --- /dev/null +++ b/ydb_tech/topic/async_example.py @@ -0,0 +1,316 @@ +"""Executable source of the asynchronous topic snippets on ydb.tech.""" + +import asyncio +import collections +import datetime +import os +import uuid + +import ydb +import ydb.aio + +EXPECTED = [ + b"text", + bytes([1, 2, 3]), + b"batch-one", + b"batch-two", + b"manual-one", + b"manual-two", + b"buffer-one", + b"buffer-two", + b"ack-batch-one", + b"ack-batch-two", + b"ack-one", + b"metadata", + b"compressed", +] + + +async def run(): + endpoint = os.getenv("YDB_ENDPOINT", "grpc://localhost:2136") + database = os.getenv("YDB_DATABASE", "/local") + topic_path = "ydb_tech_" + uuid.uuid4().hex + consumers = ["one", "batch", "commit_one", "commit_batch", "soft", "hard", "outside", "transaction"] + # [BEGIN topic_init] + config = ydb.DriverConfig(endpoint=endpoint, database=database) + async with ydb.aio.Driver(config) as driver: + await driver.wait(timeout=30) + # [END topic_init] + # [BEGIN topic_create] + await driver.topic_client.create_topic( + topic_path, + consumers=consumers, + supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], + min_active_partitions=3, + max_active_partitions=3, + ) + # [END topic_create] + try: + # [BEGIN topic_alter] + await driver.topic_client.alter_topic( + topic_path, + set_supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], + set_min_active_partitions=3, + add_consumers=["another-consumer"], + ) + # [END topic_alter] + # [BEGIN topic_describe] + info = await driver.topic_client.describe_topic(topic_path) + print("Consumers:", [consumer.name for consumer in info.consumers]) + # [END topic_describe] + assert len(info.consumers) == len(consumers) + 1 + + await seed(driver, topic_path) + for consumer in ["one", "batch", "commit_one", "commit_batch", "soft"]: + await consume(driver, topic_path, consumer) + await without_consumer(driver, topic_path) + await expired_batch(driver, topic_path) + await commit_outside(driver, topic_path) + await transactions(driver, topic_path) + await autoscaling(driver, topic_path + "_auto") + finally: + # [BEGIN topic_drop] + await driver.topic_client.drop_topic(topic_path) + # [END topic_drop] + print("All asynchronous topic scenarios completed") + + +async def seed(driver, topic_path): + # [BEGIN topic_start_writer] + writer = driver.topic_client.writer(topic_path, partition_id=0) + # [END topic_start_writer] + async with writer: + # [BEGIN topic_write] + await writer.write("text") + await writer.write(bytes([1, 2, 3])) + await writer.write(["batch-one", "batch-two"]) + # [END topic_write] + # [BEGIN topic_write_ack] + for message in ["buffer-one", "buffer-two"]: + await writer.write(message) + await writer.flush() + await writer.write_with_ack( + ["ack-batch-one", "ack-batch-two"], + ) + await writer.write_with_ack( + "ack-one", + ) + # [END topic_write_ack] + # [BEGIN topic_write_metadata] + message = ydb.TopicWriterMessage(data="metadata", metadata_items={"meta-key": "meta-value"}) + await writer.write(message) + # [END topic_write_metadata] + + # [BEGIN topic_write_manual] + async with driver.topic_client.writer( + topic_path, partition_id=0, auto_seqno=False, auto_created_at=False + ) as writer: + await writer.write( + [ + ydb.TopicWriterMessage( + "manual-one", seqno=123, created_at=datetime.datetime.now(datetime.timezone.utc) + ), + ydb.TopicWriterMessage( + "manual-two", seqno=124, created_at=datetime.datetime.now(datetime.timezone.utc) + ), + ] + ) + # [END topic_write_manual] + # [BEGIN topic_codec] + async with driver.topic_client.writer(topic_path, partition_id=0, codec=ydb.TopicCodec.GZIP) as writer: + await writer.write_with_ack( + "compressed", + ) + # [END topic_codec] + + +async def consume(driver, topic_path, consumer_name): + # [BEGIN topic_start_reader] + reader = driver.topic_client.reader(topic=topic_path, consumer=consumer_name) + # [END topic_start_reader] + received = [] + async with reader: + if consumer_name == "one": + # [BEGIN topic_read_one] + while len(received) < len(EXPECTED): + message = await asyncio.wait_for(reader.receive_message(), timeout=30) + received.append(check_message(message)) + # [END topic_read_one] + elif consumer_name == "batch": + # [BEGIN topic_read_batch] + while len(received) < len(EXPECTED): + batch = await asyncio.wait_for(reader.receive_batch(), timeout=30) + received.extend(check_message(message) for message in batch.messages) + # [END topic_read_batch] + elif consumer_name == "commit_one": + # [BEGIN topic_read_commit] + while len(received) < len(EXPECTED): + message = await asyncio.wait_for(reader.receive_message(), timeout=30) + received.append(check_message(message)) + reader.commit(message) + # [END topic_read_commit] + elif consumer_name == "commit_batch": + # [BEGIN topic_read_batch_commit] + while len(received) < len(EXPECTED): + batch = await asyncio.wait_for(reader.receive_batch(), timeout=30) + received.extend(check_message(message) for message in batch.messages) + reader.commit(batch) + # [END topic_read_batch_commit] + else: + # [BEGIN topic_soft_stop] + while len(received) < len(EXPECTED): + batch = await asyncio.wait_for(reader.receive_batch(), timeout=30) + received.extend(check_message(message) for message in batch.messages) + reader.commit(batch) + # [END topic_soft_stop] + assert collections.Counter(received) == collections.Counter(EXPECTED) + + +def check_message(message): + assert message is not None + assert message.data in EXPECTED + if message.data == b"metadata": + # [BEGIN topic_read_metadata] + for key, value in message.metadata_items.items(): + print(key, value) + # [END topic_read_metadata] + assert message.metadata_items["meta-key"] == b"meta-value" + return message.data + + +# [BEGIN topic_no_consumer_handler] +class StartAtBeginning(ydb.TopicReaderEvents.EventHandler): + def on_partition_get_start_offset(self, event): + return ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse(start_offset=0) + + +# [END topic_no_consumer_handler] + + +async def without_consumer(driver, topic_path): + # [BEGIN topic_no_consumer] + async with driver.topic_client.reader( + topic=ydb.TopicReaderSelector(path=topic_path, partitions=[0]), + consumer=None, + event_handler=StartAtBeginning(), + ) as reader: + message = await asyncio.wait_for(reader.receive_message(), timeout=30) + check_message(message) + # [END topic_no_consumer] + + +async def expired_batch(driver, topic_path): + reader = driver.topic_client.reader(topic=topic_path, consumer="hard") + try: + batch = await asyncio.wait_for(reader.receive_batch(max_messages=1), timeout=30) + assert process_batch(batch) + finally: + await reader.close(flush=False) + assert not process_batch(batch) + + +# [BEGIN topic_hard_stop] +def process_batch(batch): + for message in batch.messages: + if not batch.alive: + return False + check_message(message) + return True + + +# [END topic_hard_stop] + + +async def commit_outside(driver, topic_path): + async with driver.topic_client.reader(topic_path, "outside") as reader: + message = await asyncio.wait_for(reader.receive_message(), timeout=30) + check_message(message) + # [BEGIN topic_commit_outside] + await driver.topic_client.commit_offset( + topic_path, + "outside", + 0, + message.offset + 1, + reader.read_session_id, + ) + # [END topic_commit_outside] + + +async def transactions(driver, topic_path): + transaction_topic = topic_path + "_tx" + await driver.topic_client.create_topic(transaction_topic, consumers=["transaction"]) + try: + # [BEGIN topic_write_tx] + async with ydb.aio.QuerySessionPool(driver) as pool: + + async def write_in_transaction(tx): + writer = driver.topic_client.tx_writer(tx, transaction_topic) + async with await tx.execute("SELECT 'transaction-message' AS payload") as results: + async for result_set in results: + payload = result_set.rows[0]["payload"] + await writer.write(ydb.TopicWriterMessage(payload)) + + await pool.retry_tx_async(write_in_transaction) + # [END topic_write_tx] + + # [BEGIN topic_read_tx] + async with driver.topic_client.reader(transaction_topic, "transaction") as reader: + async with ydb.aio.QuerySessionPool(driver) as pool: + + async def read_in_transaction(tx): + batch = await asyncio.wait_for(reader.receive_batch_with_tx(tx, max_messages=1), timeout=30) + assert batch.messages[0].data == b"transaction-message" + + await pool.retry_tx_async(read_in_transaction) + # [END topic_read_tx] + finally: + await driver.topic_client.drop_topic(transaction_topic) + + +async def autoscaling(driver, topic_path): + # [BEGIN topic_autoscale_create] + await driver.topic_client.create_topic( + topic_path, + consumers=["auto"], + min_active_partitions=1, + max_active_partitions=4, + auto_partitioning_settings=ydb.TopicAutoPartitioningSettings( + strategy=ydb.TopicAutoPartitioningStrategy.SCALE_UP, + up_utilization_percent=80, + down_utilization_percent=20, + stabilization_window=datetime.timedelta(seconds=300), + ), + ) + # [END topic_autoscale_create] + try: + # [BEGIN topic_autoscale_alter] + await driver.topic_client.alter_topic( + topic_path, + alter_auto_partitioning_settings=ydb.TopicAlterAutoPartitioningSettings( + set_strategy=ydb.TopicAutoPartitioningStrategy.SCALE_UP, + set_up_utilization_percent=80, + set_down_utilization_percent=20, + set_stabilization_window=datetime.timedelta(seconds=300), + ), + ) + # [END topic_autoscale_alter] + async with driver.topic_client.writer(topic_path) as writer: + await writer.write_with_ack( + "auto", + ) + # [BEGIN topic_autoscale_reader] + for full_support in [True, False]: + async with driver.topic_client.reader( + topic_path, + "auto", + auto_partitioning_support=full_support, + ) as reader: + message = await asyncio.wait_for(reader.receive_message(), timeout=30) + assert message.data == b"auto" + # [END topic_autoscale_reader] + finally: + await driver.topic_client.drop_topic(topic_path) + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/ydb_tech/topic/run.sh b/ydb_tech/topic/run.sh new file mode 100755 index 00000000..48c7d516 --- /dev/null +++ b/ydb_tech/topic/run.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +set -euo pipefail +cd "$(dirname "$0")/../.." +python ydb_tech/topic/sync_example.py +python ydb_tech/topic/async_example.py diff --git a/ydb_tech/topic/sync_example.py b/ydb_tech/topic/sync_example.py new file mode 100644 index 00000000..678df727 --- /dev/null +++ b/ydb_tech/topic/sync_example.py @@ -0,0 +1,303 @@ +"""Executable source of the synchronous topic snippets on ydb.tech.""" + +import collections +import datetime +import os +import uuid + +import ydb + +EXPECTED = [ + b"text", + bytes([1, 2, 3]), + b"batch-one", + b"batch-two", + b"manual-one", + b"manual-two", + b"buffer-one", + b"buffer-two", + b"ack-batch-one", + b"ack-batch-two", + b"ack-one", + b"metadata", + b"compressed", +] + + +def run(): + endpoint = os.getenv("YDB_ENDPOINT", "grpc://localhost:2136") + database = os.getenv("YDB_DATABASE", "/local") + topic_path = "ydb_tech_" + uuid.uuid4().hex + consumers = ["one", "batch", "commit_one", "commit_batch", "soft", "hard", "outside", "transaction"] + # [BEGIN topic_init] + config = ydb.DriverConfig(endpoint=endpoint, database=database) + with ydb.Driver(config) as driver: + driver.wait(timeout=30) + # [END topic_init] + # [BEGIN topic_create] + driver.topic_client.create_topic( + topic_path, + consumers=consumers, + supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], + min_active_partitions=3, + max_active_partitions=3, + ) + # [END topic_create] + try: + # [BEGIN topic_alter] + driver.topic_client.alter_topic( + topic_path, + set_supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], + set_min_active_partitions=3, + add_consumers=["another-consumer"], + ) + # [END topic_alter] + # [BEGIN topic_describe] + info = driver.topic_client.describe_topic(topic_path) + print("Consumers:", [consumer.name for consumer in info.consumers]) + # [END topic_describe] + assert len(info.consumers) == len(consumers) + 1 + + seed(driver, topic_path) + for consumer in ["one", "batch", "commit_one", "commit_batch", "soft"]: + consume(driver, topic_path, consumer) + without_consumer(driver, topic_path) + expired_batch(driver, topic_path) + commit_outside(driver, topic_path) + transactions(driver, topic_path) + autoscaling(driver, topic_path + "_auto") + finally: + # [BEGIN topic_drop] + driver.topic_client.drop_topic(topic_path) + # [END topic_drop] + print("All synchronous topic scenarios completed") + + +def seed(driver, topic_path): + # [BEGIN topic_start_writer] + writer = driver.topic_client.writer(topic_path, partition_id=0) + # [END topic_start_writer] + with writer: + # [BEGIN topic_write] + writer.write("text") + writer.write(bytes([1, 2, 3])) + writer.write(["batch-one", "batch-two"]) + # [END topic_write] + # [BEGIN topic_write_ack] + for message in ["buffer-one", "buffer-two"]: + writer.write(message) + writer.flush(timeout=30) + writer.write_with_ack(["ack-batch-one", "ack-batch-two"], timeout=30) + writer.write_with_ack("ack-one", timeout=30) + # [END topic_write_ack] + # [BEGIN topic_write_metadata] + message = ydb.TopicWriterMessage(data="metadata", metadata_items={"meta-key": "meta-value"}) + writer.write(message) + # [END topic_write_metadata] + + # [BEGIN topic_write_manual] + with driver.topic_client.writer(topic_path, partition_id=0, auto_seqno=False, auto_created_at=False) as writer: + writer.write( + [ + ydb.TopicWriterMessage( + "manual-one", seqno=123, created_at=datetime.datetime.now(datetime.timezone.utc) + ), + ydb.TopicWriterMessage( + "manual-two", seqno=124, created_at=datetime.datetime.now(datetime.timezone.utc) + ), + ] + ) + # [END topic_write_manual] + # [BEGIN topic_codec] + with driver.topic_client.writer(topic_path, partition_id=0, codec=ydb.TopicCodec.GZIP) as writer: + writer.write_with_ack("compressed", timeout=30) + # [END topic_codec] + + +def consume(driver, topic_path, consumer_name): + # [BEGIN topic_start_reader] + reader = driver.topic_client.reader(topic=topic_path, consumer=consumer_name) + # [END topic_start_reader] + received = [] + with reader: + if consumer_name == "one": + # [BEGIN topic_read_one] + while len(received) < len(EXPECTED): + message = reader.receive_message(timeout=30) + received.append(check_message(message)) + # [END topic_read_one] + elif consumer_name == "batch": + # [BEGIN topic_read_batch] + while len(received) < len(EXPECTED): + batch = reader.receive_batch(timeout=30) + received.extend(check_message(message) for message in batch.messages) + # [END topic_read_batch] + elif consumer_name == "commit_one": + # [BEGIN topic_read_commit] + while len(received) < len(EXPECTED): + message = reader.receive_message(timeout=30) + received.append(check_message(message)) + reader.commit(message) + # [END topic_read_commit] + elif consumer_name == "commit_batch": + # [BEGIN topic_read_batch_commit] + while len(received) < len(EXPECTED): + batch = reader.receive_batch(timeout=30) + received.extend(check_message(message) for message in batch.messages) + reader.commit(batch) + # [END topic_read_batch_commit] + else: + # [BEGIN topic_soft_stop] + while len(received) < len(EXPECTED): + batch = reader.receive_batch(timeout=30) + received.extend(check_message(message) for message in batch.messages) + reader.commit(batch) + # [END topic_soft_stop] + assert collections.Counter(received) == collections.Counter(EXPECTED) + + +def check_message(message): + assert message is not None + assert message.data in EXPECTED + if message.data == b"metadata": + # [BEGIN topic_read_metadata] + for key, value in message.metadata_items.items(): + print(key, value) + # [END topic_read_metadata] + assert message.metadata_items["meta-key"] == b"meta-value" + return message.data + + +# [BEGIN topic_no_consumer_handler] +class StartAtBeginning(ydb.TopicReaderEvents.EventHandler): + def on_partition_get_start_offset(self, event): + return ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse(start_offset=0) + + +# [END topic_no_consumer_handler] + + +def without_consumer(driver, topic_path): + # [BEGIN topic_no_consumer] + with driver.topic_client.reader( + topic=ydb.TopicReaderSelector(path=topic_path, partitions=[0]), + consumer=None, + event_handler=StartAtBeginning(), + ) as reader: + message = reader.receive_message(timeout=30) + check_message(message) + # [END topic_no_consumer] + + +def expired_batch(driver, topic_path): + reader = driver.topic_client.reader(topic=topic_path, consumer="hard") + try: + batch = reader.receive_batch(max_messages=1, timeout=30) + assert process_batch(batch) + finally: + reader.close(flush=False, timeout=30) + assert not process_batch(batch) + + +# [BEGIN topic_hard_stop] +def process_batch(batch): + for message in batch.messages: + if not batch.alive: + return False + check_message(message) + return True + + +# [END topic_hard_stop] + + +def commit_outside(driver, topic_path): + with driver.topic_client.reader(topic_path, "outside") as reader: + message = reader.receive_message(timeout=30) + check_message(message) + # [BEGIN topic_commit_outside] + driver.topic_client.commit_offset( + topic_path, + "outside", + 0, + message.offset + 1, + reader.read_session_id, + ) + # [END topic_commit_outside] + + +def transactions(driver, topic_path): + transaction_topic = topic_path + "_tx" + driver.topic_client.create_topic(transaction_topic, consumers=["transaction"]) + try: + # [BEGIN topic_write_tx] + with ydb.QuerySessionPool(driver) as pool: + + def write_in_transaction(tx): + writer = driver.topic_client.tx_writer(tx, transaction_topic) + for result_set in tx.execute("SELECT 'transaction-message' AS payload"): + payload = result_set.rows[0]["payload"] + writer.write(ydb.TopicWriterMessage(payload)) + + pool.retry_tx_sync(write_in_transaction) + # [END topic_write_tx] + + # [BEGIN topic_read_tx] + with driver.topic_client.reader(transaction_topic, "transaction") as reader: + with ydb.QuerySessionPool(driver) as pool: + + def read_in_transaction(tx): + batch = reader.receive_batch_with_tx(tx, max_messages=1, timeout=30) + assert batch.messages[0].data == b"transaction-message" + + pool.retry_tx_sync(read_in_transaction) + # [END topic_read_tx] + finally: + driver.topic_client.drop_topic(transaction_topic) + + +def autoscaling(driver, topic_path): + # [BEGIN topic_autoscale_create] + driver.topic_client.create_topic( + topic_path, + consumers=["auto"], + min_active_partitions=1, + max_active_partitions=4, + auto_partitioning_settings=ydb.TopicAutoPartitioningSettings( + strategy=ydb.TopicAutoPartitioningStrategy.SCALE_UP, + up_utilization_percent=80, + down_utilization_percent=20, + stabilization_window=datetime.timedelta(seconds=300), + ), + ) + # [END topic_autoscale_create] + try: + # [BEGIN topic_autoscale_alter] + driver.topic_client.alter_topic( + topic_path, + alter_auto_partitioning_settings=ydb.TopicAlterAutoPartitioningSettings( + set_strategy=ydb.TopicAutoPartitioningStrategy.SCALE_UP, + set_up_utilization_percent=80, + set_down_utilization_percent=20, + set_stabilization_window=datetime.timedelta(seconds=300), + ), + ) + # [END topic_autoscale_alter] + with driver.topic_client.writer(topic_path) as writer: + writer.write_with_ack("auto", timeout=30) + # [BEGIN topic_autoscale_reader] + for full_support in [True, False]: + with driver.topic_client.reader( + topic_path, + "auto", + auto_partitioning_support=full_support, + ) as reader: + message = reader.receive_message(timeout=30) + assert message.data == b"auto" + # [END topic_autoscale_reader] + finally: + driver.topic_client.drop_topic(topic_path) + + +if __name__ == "__main__": + run() From 3138cc3ccee2bfbd5aef08329f34a3c47add2598 Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Tue, 6 Oct 2026 12:22:13 +0300 Subject: [PATCH 2/7] Move documentation examples under examples --- .github/workflows/ydb-tech.yml | 2 +- {ydb_tech => examples/ydb_tech}/README.md | 2 +- {ydb_tech => examples/ydb_tech}/run.sh | 2 +- {ydb_tech => examples/ydb_tech}/topic/README.md | 4 ++-- {ydb_tech => examples/ydb_tech}/topic/async_example.py | 0 examples/ydb_tech/topic/run.sh | 5 +++++ {ydb_tech => examples/ydb_tech}/topic/sync_example.py | 0 ydb_tech/topic/run.sh | 5 ----- 8 files changed, 10 insertions(+), 10 deletions(-) rename {ydb_tech => examples/ydb_tech}/README.md (97%) rename {ydb_tech => examples/ydb_tech}/run.sh (92%) rename {ydb_tech => examples/ydb_tech}/topic/README.md (85%) rename {ydb_tech => examples/ydb_tech}/topic/async_example.py (100%) create mode 100755 examples/ydb_tech/topic/run.sh rename {ydb_tech => examples/ydb_tech}/topic/sync_example.py (100%) delete mode 100755 ydb_tech/topic/run.sh diff --git a/.github/workflows/ydb-tech.yml b/.github/workflows/ydb-tech.yml index f8b2fcba..aefe0ac5 100644 --- a/.github/workflows/ydb-tech.yml +++ b/.github/workflows/ydb-tech.yml @@ -30,4 +30,4 @@ jobs: with: python-version: '3.10' - run: python -m pip install -e . - - run: bash ydb_tech/run.sh + - run: bash examples/ydb_tech/run.sh diff --git a/ydb_tech/README.md b/examples/ydb_tech/README.md similarity index 97% rename from ydb_tech/README.md rename to examples/ydb_tech/README.md index 62e80e87..4ca70dbe 100644 --- a/ydb_tech/README.md +++ b/examples/ydb_tech/README.md @@ -7,7 +7,7 @@ this checkout and a local YDB instance. Run all examples from the repository root: ```sh -bash ydb_tech/run.sh +bash examples/ydb_tech/run.sh ``` Each example has its own directory and a `run.sh` entry point. The common runner diff --git a/ydb_tech/run.sh b/examples/ydb_tech/run.sh similarity index 92% rename from ydb_tech/run.sh rename to examples/ydb_tech/run.sh index 1b4bdcbb..5b5efcf9 100755 --- a/ydb_tech/run.sh +++ b/examples/ydb_tech/run.sh @@ -1,7 +1,7 @@ #!/usr/bin/env bash set -euo pipefail examples_root="$(cd "$(dirname "$0")" && pwd)" -repo_root="$(dirname "$examples_root")" +repo_root="$(git -C "$examples_root" rev-parse --show-toplevel)" cd "$repo_root" example_count=0 for entry in "$examples_root"/*/run.sh; do diff --git a/ydb_tech/topic/README.md b/examples/ydb_tech/topic/README.md similarity index 85% rename from ydb_tech/topic/README.md rename to examples/ydb_tech/topic/README.md index 9d80b064..32c7b90e 100644 --- a/ydb_tech/topic/README.md +++ b/examples/ydb_tech/topic/README.md @@ -7,8 +7,8 @@ From the repository root, with a local YDB instance: ```sh python -m pip install -e . -python ydb_tech/topic/sync_example.py -python ydb_tech/topic/async_example.py +python examples/ydb_tech/topic/sync_example.py +python examples/ydb_tech/topic/async_example.py ``` `YDB_ENDPOINT` defaults to `grpc://localhost:2136` and `YDB_DATABASE` to `/local`. diff --git a/ydb_tech/topic/async_example.py b/examples/ydb_tech/topic/async_example.py similarity index 100% rename from ydb_tech/topic/async_example.py rename to examples/ydb_tech/topic/async_example.py diff --git a/examples/ydb_tech/topic/run.sh b/examples/ydb_tech/topic/run.sh new file mode 100755 index 00000000..146751d5 --- /dev/null +++ b/examples/ydb_tech/topic/run.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +set -euo pipefail +cd "$(git -C "$(dirname "$0")" rev-parse --show-toplevel)" +python examples/ydb_tech/topic/sync_example.py +python examples/ydb_tech/topic/async_example.py diff --git a/ydb_tech/topic/sync_example.py b/examples/ydb_tech/topic/sync_example.py similarity index 100% rename from ydb_tech/topic/sync_example.py rename to examples/ydb_tech/topic/sync_example.py diff --git a/ydb_tech/topic/run.sh b/ydb_tech/topic/run.sh deleted file mode 100755 index 48c7d516..00000000 --- a/ydb_tech/topic/run.sh +++ /dev/null @@ -1,5 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail -cd "$(dirname "$0")/../.." -python ydb_tech/topic/sync_example.py -python ydb_tech/topic/async_example.py From b1e746a4b94df3d4040c37cb3026d25822388bce Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Thu, 8 Oct 2026 14:14:12 +0300 Subject: [PATCH 3/7] Restore documented Python topic snippets --- examples/ydb_tech/topic/async_example.py | 440 +++++++++++++--------- examples/ydb_tech/topic/sync_example.py | 453 ++++++++++++++--------- 2 files changed, 545 insertions(+), 348 deletions(-) diff --git a/examples/ydb_tech/topic/async_example.py b/examples/ydb_tech/topic/async_example.py index d0a2c655..61a3a19c 100644 --- a/examples/ydb_tech/topic/async_example.py +++ b/examples/ydb_tech/topic/async_example.py @@ -10,118 +10,162 @@ import ydb.aio EXPECTED = [ - b"text", + b"mess", bytes([1, 2, 3]), - b"batch-one", - b"batch-two", - b"manual-one", - b"manual-two", + b"mess-1", + b"mess-2", b"buffer-one", b"buffer-two", - b"ack-batch-one", - b"ack-batch-two", - b"ack-one", - b"metadata", + b"mess-1", + b"mess-2", + b"message", + b"message-data", + b"asd", + bytes([1, 2, 3]), b"compressed", ] +class ScenarioComplete(Exception): + pass + + +class FiniteReader: + def __init__(self, reader, received): + self.reader = reader + self.received = received + + def __getattr__(self, name): + return getattr(self.reader, name) + + async def receive_message(self): + if len(self.received) >= len(EXPECTED): + raise ScenarioComplete + return await asyncio.wait_for(self.reader.receive_message(), 30) + + async def receive_batch(self): + if len(self.received) >= len(EXPECTED): + raise ScenarioComplete + return await asyncio.wait_for(self.reader.receive_batch(), 30) + + +def check_message(message): + assert message.data in EXPECTED + if message.data == b"message-data": + assert message.metadata_items["meta-key"] == b"meta-value" + return message.data + + +async def initialize(topic_path, consumer_name): + # [BEGIN topic_init] + import os + import ydb + + driver_config = ydb.DriverConfig( + endpoint=os.environ["YDB_ENDPOINT"], + database=os.environ["YDB_DATABASE"], + ) + async with ydb.aio.Driver(driver_config) as driver: + await driver.wait(timeout=5) + # driver.topic_client — client for working with topics + writer = driver.topic_client.writer(topic_path) + reader = driver.topic_client.reader(topic=topic_path, consumer=consumer_name) + # [END topic_init] + await writer.close() + await reader.close() + + async def run(): - endpoint = os.getenv("YDB_ENDPOINT", "grpc://localhost:2136") - database = os.getenv("YDB_DATABASE", "/local") + os.environ.setdefault("YDB_ENDPOINT", "grpc://localhost:2136") + os.environ.setdefault("YDB_DATABASE", "/local") topic_path = "ydb_tech_" + uuid.uuid4().hex - consumers = ["one", "batch", "commit_one", "commit_batch", "soft", "hard", "outside", "transaction"] - # [BEGIN topic_init] - config = ydb.DriverConfig(endpoint=endpoint, database=database) + consumers = ["init", "one", "batch", "commit_one", "commit_batch", "soft", "hard", "outside"] + config = ydb.DriverConfig(endpoint=os.environ["YDB_ENDPOINT"], database=os.environ["YDB_DATABASE"]) async with ydb.aio.Driver(config) as driver: await driver.wait(timeout=30) - # [END topic_init] # [BEGIN topic_create] await driver.topic_client.create_topic( topic_path, - consumers=consumers, - supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], - min_active_partitions=3, - max_active_partitions=3, + supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], # optional + min_active_partitions=3, # optional ) # [END topic_create] try: + await driver.topic_client.alter_topic(topic_path, add_consumers=consumers) + await initialize(topic_path, "init") # [BEGIN topic_alter] await driver.topic_client.alter_topic( topic_path, - set_supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], - set_min_active_partitions=3, - add_consumers=["another-consumer"], + set_supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], # optional + set_min_active_partitions=3, # optional ) # [END topic_alter] # [BEGIN topic_describe] info = await driver.topic_client.describe_topic(topic_path) - print("Consumers:", [consumer.name for consumer in info.consumers]) + print(info) # [END topic_describe] - assert len(info.consumers) == len(consumers) + 1 - + assert len(info.consumers) == len(consumers) await seed(driver, topic_path) for consumer in ["one", "batch", "commit_one", "commit_batch", "soft"]: await consume(driver, topic_path, consumer) + await metadata(driver, topic_path + "_metadata") await without_consumer(driver, topic_path) await expired_batch(driver, topic_path) await commit_outside(driver, topic_path) - await transactions(driver, topic_path) + await transactions(driver, topic_path + "_tx") await autoscaling(driver, topic_path + "_auto") finally: # [BEGIN topic_drop] await driver.topic_client.drop_topic(topic_path) # [END topic_drop] - print("All asynchronous topic scenarios completed") -async def seed(driver, topic_path): +async def start_writer(driver, topic_path): # [BEGIN topic_start_writer] - writer = driver.topic_client.writer(topic_path, partition_id=0) + writer = driver.topic_client.writer(topic_path) # [END topic_start_writer] - async with writer: - # [BEGIN topic_write] - await writer.write("text") - await writer.write(bytes([1, 2, 3])) - await writer.write(["batch-one", "batch-two"]) - # [END topic_write] + return writer + + +async def seed(driver, topic_path): + messages = ["buffer-one", "buffer-two"] + # [BEGIN topic_write] + writer = driver.topic_client.writer(topic_path) + await writer.write("mess") + await writer.write(bytes([1, 2, 3])) + await writer.write(["mess-1", "mess-2"]) + # [END topic_write] + try: # [BEGIN topic_write_ack] - for message in ["buffer-one", "buffer-two"]: - await writer.write(message) + for mess in messages: + await writer.write(mess) + await writer.flush() - await writer.write_with_ack( - ["ack-batch-one", "ack-batch-two"], - ) - await writer.write_with_ack( - "ack-one", - ) + + await writer.write_with_ack(["mess-1", "mess-2"]) + await writer.write_with_ack("message") # [END topic_write_ack] # [BEGIN topic_write_metadata] - message = ydb.TopicWriterMessage(data="metadata", metadata_items={"meta-key": "meta-value"}) + message = ydb.TopicWriterMessage(data="message-data", metadata_items={"meta-key": "meta-value"}) await writer.write(message) # [END topic_write_metadata] - - # [BEGIN topic_write_manual] - async with driver.topic_client.writer( - topic_path, partition_id=0, auto_seqno=False, auto_created_at=False - ) as writer: + finally: + await writer.close() + writer = driver.topic_client.writer(topic_path, auto_seqno=False, auto_created_at=False) + async with writer: await writer.write( [ - ydb.TopicWriterMessage( - "manual-one", seqno=123, created_at=datetime.datetime.now(datetime.timezone.utc) - ), - ydb.TopicWriterMessage( - "manual-two", seqno=124, created_at=datetime.datetime.now(datetime.timezone.utc) - ), + ydb.TopicWriterMessage("asd", seqno=123, created_at=datetime.datetime.now()), + ydb.TopicWriterMessage(bytes([1, 2, 3]), seqno=124, created_at=datetime.datetime.now()), ] ) - # [END topic_write_manual] - # [BEGIN topic_codec] - async with driver.topic_client.writer(topic_path, partition_id=0, codec=ydb.TopicCodec.GZIP) as writer: - await writer.write_with_ack( - "compressed", - ) - # [END topic_codec] + # codec setup + writer = driver.topic_client.writer( + topic_path, + codec=ydb.TopicCodec.GZIP, + ) + async with writer: + await writer.write_with_ack("compressed") async def consume(driver, topic_path, consumer_name): @@ -129,151 +173,178 @@ async def consume(driver, topic_path, consumer_name): reader = driver.topic_client.reader(topic=topic_path, consumer=consumer_name) # [END topic_start_reader] received = [] - async with reader: - if consumer_name == "one": - # [BEGIN topic_read_one] - while len(received) < len(EXPECTED): - message = await asyncio.wait_for(reader.receive_message(), timeout=30) - received.append(check_message(message)) - # [END topic_read_one] - elif consumer_name == "batch": - # [BEGIN topic_read_batch] - while len(received) < len(EXPECTED): - batch = await asyncio.wait_for(reader.receive_batch(), timeout=30) - received.extend(check_message(message) for message in batch.messages) - # [END topic_read_batch] - elif consumer_name == "commit_one": - # [BEGIN topic_read_commit] - while len(received) < len(EXPECTED): - message = await asyncio.wait_for(reader.receive_message(), timeout=30) - received.append(check_message(message)) - reader.commit(message) - # [END topic_read_commit] - elif consumer_name == "commit_batch": - # [BEGIN topic_read_batch_commit] - while len(received) < len(EXPECTED): - batch = await asyncio.wait_for(reader.receive_batch(), timeout=30) - received.extend(check_message(message) for message in batch.messages) - reader.commit(batch) - # [END topic_read_batch_commit] - else: - # [BEGIN topic_soft_stop] - while len(received) < len(EXPECTED): - batch = await asyncio.wait_for(reader.receive_batch(), timeout=30) - received.extend(check_message(message) for message in batch.messages) - reader.commit(batch) - # [END topic_soft_stop] - assert collections.Counter(received) == collections.Counter(EXPECTED) - - -def check_message(message): - assert message is not None - assert message.data in EXPECTED - if message.data == b"metadata": - # [BEGIN topic_read_metadata] - for key, value in message.metadata_items.items(): - print(key, value) - # [END topic_read_metadata] - assert message.metadata_items["meta-key"] == b"meta-value" - return message.data + def process(value): + messages = value.messages if hasattr(value, "messages") else [value] + received.extend(check_message(message) for message in messages) -# [BEGIN topic_no_consumer_handler] -class StartAtBeginning(ydb.TopicReaderEvents.EventHandler): - def on_partition_get_start_offset(self, event): - return ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse(start_offset=0) + async with reader: + reader = FiniteReader(reader, received) + try: + if consumer_name == "one": + # [BEGIN topic_read_one] + while True: + message = await reader.receive_message() + process(message) + # [END topic_read_one] + elif consumer_name == "batch": + # [BEGIN topic_read_batch] + while True: + batch = await reader.receive_batch() + process(batch) + # [END topic_read_batch] + elif consumer_name == "commit_one": + # [BEGIN topic_read_commit] + while True: + message = await reader.receive_message() + process(message) + reader.commit(message) + # [END topic_read_commit] + elif consumer_name == "commit_batch": + # [BEGIN topic_read_batch_commit] + while True: + batch = await reader.receive_batch() + process(batch) + reader.commit(batch) + # [END topic_read_batch_commit] + elif consumer_name == "soft": + # [BEGIN topic_soft_stop] + while True: + batch = await reader.receive_batch() + process(batch) + reader.commit(batch) + # [END topic_soft_stop] + except ScenarioComplete: + pass + assert collections.Counter(received) == collections.Counter(EXPECTED) + + +async def metadata(driver, topic_path): + await driver.topic_client.create_topic(topic_path, consumers=["metadata"]) + try: + async with driver.topic_client.writer(topic_path) as writer: + await writer.write_with_ack( + ydb.TopicWriterMessage("message-data", metadata_items={"meta-key": "meta-value"}) + ) + async with driver.topic_client.reader(topic_path, "metadata") as reader: + message = await reader.receive_message() + for meta_key, meta_value in message.metadata_items.items(): + print(f"{meta_key}: {meta_value}") + assert message.metadata_items["meta-key"] == b"meta-value" + finally: + await driver.topic_client.drop_topic(topic_path) -# [END topic_no_consumer_handler] +# Consumerless reader handler +class CustomEventHandler(ydb.TopicReaderEvents.EventHandler): + def on_partition_get_start_offset(self, event: ydb.TopicReaderEvents.OnPartitionGetStartOffsetRequest): + return ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse( + start_offset=0, + ) async def without_consumer(driver, topic_path): - # [BEGIN topic_no_consumer] - async with driver.topic_client.reader( - topic=ydb.TopicReaderSelector(path=topic_path, partitions=[0]), + reader = driver.topic_client.reader( + topic=ydb.TopicReaderSelector( + path=topic_path, + partitions=[0, 1, 2], + ), consumer=None, - event_handler=StartAtBeginning(), - ) as reader: - message = await asyncio.wait_for(reader.receive_message(), timeout=30) + event_handler=CustomEventHandler(), + ) + async with reader: + message = await asyncio.wait_for(reader.receive_message(), 30) check_message(message) - # [END topic_no_consumer] async def expired_batch(driver, topic_path): - reader = driver.topic_client.reader(topic=topic_path, consumer="hard") + reader = driver.topic_client.reader(topic_path, "hard") + process = check_message try: - batch = await asyncio.wait_for(reader.receive_batch(max_messages=1), timeout=30) - assert process_batch(batch) + # [BEGIN topic_hard_stop] + def process_batch(batch): + for message in batch.messages: + if not batch.alive: + return False + process(message) + return True + + batch = await reader.receive_batch() + if process_batch(batch): + reader.commit(batch) + # [END topic_hard_stop] finally: await reader.close(flush=False) assert not process_batch(batch) -# [BEGIN topic_hard_stop] -def process_batch(batch): - for message in batch.messages: - if not batch.alive: - return False - check_message(message) - return True - - -# [END topic_hard_stop] - - async def commit_outside(driver, topic_path): - async with driver.topic_client.reader(topic_path, "outside") as reader: - message = await asyncio.wait_for(reader.receive_message(), timeout=30) + consumer_name = "outside" + async with driver.topic_client.reader(topic_path, consumer_name) as reader: + message = await asyncio.wait_for(reader.receive_message(), 30) check_message(message) + partition_id = message.partition_id + offset = message.offset + 1 # [BEGIN topic_commit_outside] await driver.topic_client.commit_offset( topic_path, - "outside", - 0, - message.offset + 1, - reader.read_session_id, + consumer_name, + partition_id, + offset, + reader.read_session_id, # опционально: не прерывает активную сессию чтения ) # [END topic_commit_outside] -async def transactions(driver, topic_path): - transaction_topic = topic_path + "_tx" - await driver.topic_client.create_topic(transaction_topic, consumers=["transaction"]) +async def transactions(driver, topic): + consumer = "transaction" + message_count = 3 + await driver.topic_client.create_topic(topic, consumers=[consumer, "verify"]) try: # [BEGIN topic_write_tx] - async with ydb.aio.QuerySessionPool(driver) as pool: + async with ydb.aio.QuerySessionPool(driver) as session_pool: - async def write_in_transaction(tx): - writer = driver.topic_client.tx_writer(tx, transaction_topic) - async with await tx.execute("SELECT 'transaction-message' AS payload") as results: - async for result_set in results: - payload = result_set.rows[0]["payload"] - await writer.write(ydb.TopicWriterMessage(payload)) + async def callee(tx: ydb.aio.QueryTxContext): + tx_writer: ydb.TopicTxWriterAsyncIO = driver.topic_client.tx_writer(tx, topic) - await pool.retry_tx_async(write_in_transaction) - # [END topic_write_tx] + for i in range(message_count): + async with await tx.execute(query=f"select {i} as res;") as result_stream: + async for result_set in result_stream: + message = str(result_set.rows[0]["res"]) + await tx_writer.write(ydb.TopicWriterMessage(message)) + print(f"Message {result_set.rows[0]['res']} was written with tx.") + await session_pool.retry_tx_async(callee) + # [END topic_write_tx] + async with driver.topic_client.reader(topic, "verify") as verifier: + payloads = [] + while len(payloads) < message_count: + batch = await asyncio.wait_for(verifier.receive_batch(), 30) + payloads.extend(message.data for message in batch.messages) + assert payloads == [b"0", b"1", b"2"] # [BEGIN topic_read_tx] - async with driver.topic_client.reader(transaction_topic, "transaction") as reader: - async with ydb.aio.QuerySessionPool(driver) as pool: + async with driver.topic_client.reader(topic, consumer) as reader: + async with ydb.aio.QuerySessionPool(driver) as session_pool: + for _ in range(message_count): - async def read_in_transaction(tx): - batch = await asyncio.wait_for(reader.receive_batch_with_tx(tx, max_messages=1), timeout=30) - assert batch.messages[0].data == b"transaction-message" + async def callee(tx: ydb.aio.QueryTxContext): + batch = await reader.receive_batch_with_tx(tx, max_messages=1) + print(f"Message {batch.messages[0].data.decode()} was read with tx.") - await pool.retry_tx_async(read_in_transaction) + await session_pool.retry_tx_async(callee) # [END topic_read_tx] finally: - await driver.topic_client.drop_topic(transaction_topic) + await driver.topic_client.drop_topic(topic) -async def autoscaling(driver, topic_path): +async def autoscaling(driver, topic): + consumer = "auto" # [BEGIN topic_autoscale_create] await driver.topic_client.create_topic( - topic_path, - consumers=["auto"], - min_active_partitions=1, - max_active_partitions=4, + topic, + consumers=[consumer], + min_active_partitions=10, + max_active_partitions=100, auto_partitioning_settings=ydb.TopicAutoPartitioningSettings( strategy=ydb.TopicAutoPartitioningStrategy.SCALE_UP, up_utilization_percent=80, @@ -282,8 +353,8 @@ async def autoscaling(driver, topic_path): ), ) # [END topic_autoscale_create] + topic_path = topic try: - # [BEGIN topic_autoscale_alter] await driver.topic_client.alter_topic( topic_path, alter_auto_partitioning_settings=ydb.TopicAlterAutoPartitioningSettings( @@ -293,23 +364,26 @@ async def autoscaling(driver, topic_path): set_stabilization_window=datetime.timedelta(seconds=300), ), ) - # [END topic_autoscale_alter] - async with driver.topic_client.writer(topic_path) as writer: - await writer.write_with_ack( - "auto", - ) - # [BEGIN topic_autoscale_reader] - for full_support in [True, False]: - async with driver.topic_client.reader( - topic_path, - "auto", - auto_partitioning_support=full_support, - ) as reader: - message = await asyncio.wait_for(reader.receive_message(), timeout=30) - assert message.data == b"auto" - # [END topic_autoscale_reader] + async with driver.topic_client.writer(topic) as writer: + await writer.write_with_ack("auto") + reader = driver.topic_client.reader( + topic, + consumer, + auto_partitioning_support=True, # Full support is enabled + ) + async with reader: + message = await asyncio.wait_for(reader.receive_message(), 30) + assert message.data == b"auto" + reader = driver.topic_client.reader( + topic, + consumer, + auto_partitioning_support=False, # Compatibility mode is enabled + ) + async with reader: + message = await asyncio.wait_for(reader.receive_message(), 30) + assert message.data == b"auto" finally: - await driver.topic_client.drop_topic(topic_path) + await driver.topic_client.drop_topic(topic) if __name__ == "__main__": diff --git a/examples/ydb_tech/topic/sync_example.py b/examples/ydb_tech/topic/sync_example.py index 678df727..528875b1 100644 --- a/examples/ydb_tech/topic/sync_example.py +++ b/examples/ydb_tech/topic/sync_example.py @@ -6,112 +6,191 @@ import uuid import ydb +import ydb.aio EXPECTED = [ - b"text", + b"mess", bytes([1, 2, 3]), - b"batch-one", - b"batch-two", - b"manual-one", - b"manual-two", + b"mess-1", + b"mess-2", b"buffer-one", b"buffer-two", - b"ack-batch-one", - b"ack-batch-two", - b"ack-one", - b"metadata", + b"mess-1", + b"mess-2", + b"message", + b"message-data", + b"asd", + bytes([1, 2, 3]), b"compressed", ] +class ScenarioComplete(Exception): + pass + + +class FiniteReader: + def __init__(self, reader, received): + self.reader = reader + self.received = received + + def __getattr__(self, name): + return getattr(self.reader, name) + + def receive_message(self): + if len(self.received) >= len(EXPECTED): + raise ScenarioComplete + return self.reader.receive_message(timeout=30) + + def receive_batch(self): + if len(self.received) >= len(EXPECTED): + raise ScenarioComplete + return self.reader.receive_batch(timeout=30) + + +def check_message(message): + assert message.data in EXPECTED + if message.data == b"message-data": + assert message.metadata_items["meta-key"] == b"meta-value" + return message.data + + +def initialize(topic_path, consumer_name): + # [BEGIN topic_init] + import os + import ydb + + driver_config = ydb.DriverConfig( + endpoint=os.environ["YDB_ENDPOINT"], + database=os.environ["YDB_DATABASE"], + ) + driver = ydb.Driver(driver_config) + driver.wait(timeout=5) + # driver.topic_client — client for working with topics + writer = driver.topic_client.writer(topic_path) + reader = driver.topic_client.reader(topic=topic_path, consumer=consumer_name) + # [END topic_init] + writer.close() + reader.close() + driver.stop() + + def run(): - endpoint = os.getenv("YDB_ENDPOINT", "grpc://localhost:2136") - database = os.getenv("YDB_DATABASE", "/local") + os.environ.setdefault("YDB_ENDPOINT", "grpc://localhost:2136") + os.environ.setdefault("YDB_DATABASE", "/local") topic_path = "ydb_tech_" + uuid.uuid4().hex - consumers = ["one", "batch", "commit_one", "commit_batch", "soft", "hard", "outside", "transaction"] - # [BEGIN topic_init] - config = ydb.DriverConfig(endpoint=endpoint, database=database) + consumers = ["init", "one", "batch", "commit_one", "commit_batch", "soft", "hard", "outside"] + config = ydb.DriverConfig(endpoint=os.environ["YDB_ENDPOINT"], database=os.environ["YDB_DATABASE"]) with ydb.Driver(config) as driver: driver.wait(timeout=30) - # [END topic_init] # [BEGIN topic_create] driver.topic_client.create_topic( topic_path, - consumers=consumers, - supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], - min_active_partitions=3, - max_active_partitions=3, + supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], # optional + min_active_partitions=3, # optional ) # [END topic_create] try: + driver.topic_client.alter_topic(topic_path, add_consumers=consumers) + initialize(topic_path, "init") # [BEGIN topic_alter] driver.topic_client.alter_topic( topic_path, - set_supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], - set_min_active_partitions=3, - add_consumers=["another-consumer"], + set_supported_codecs=[ydb.TopicCodec.RAW, ydb.TopicCodec.GZIP], # optional + set_min_active_partitions=3, # optional ) # [END topic_alter] # [BEGIN topic_describe] info = driver.topic_client.describe_topic(topic_path) - print("Consumers:", [consumer.name for consumer in info.consumers]) + print(info) # [END topic_describe] - assert len(info.consumers) == len(consumers) + 1 - + assert len(info.consumers) == len(consumers) seed(driver, topic_path) for consumer in ["one", "batch", "commit_one", "commit_batch", "soft"]: consume(driver, topic_path, consumer) + metadata(driver, topic_path + "_metadata") without_consumer(driver, topic_path) expired_batch(driver, topic_path) commit_outside(driver, topic_path) - transactions(driver, topic_path) + transactions(driver, topic_path + "_tx") autoscaling(driver, topic_path + "_auto") finally: # [BEGIN topic_drop] driver.topic_client.drop_topic(topic_path) # [END topic_drop] - print("All synchronous topic scenarios completed") -def seed(driver, topic_path): +def start_writer(driver, topic_path): # [BEGIN topic_start_writer] - writer = driver.topic_client.writer(topic_path, partition_id=0) + writer = driver.topic_client.writer(topic_path) # [END topic_start_writer] - with writer: - # [BEGIN topic_write] - writer.write("text") - writer.write(bytes([1, 2, 3])) - writer.write(["batch-one", "batch-two"]) - # [END topic_write] + return writer + + +def seed(driver, topic_path): + messages = ["buffer-one", "buffer-two"] + # [BEGIN topic_write] + # Simple message sending, without explicitly specifying metadata. + # Convenient to start with, convenient to use while only the message content matters. + writer = driver.topic_client.writer(topic_path) + writer.write("mess") # Strings are encoded as utf-8; convenient for + # text messages. + writer.write(bytes([1, 2, 3])) # Bytes are sent unchanged; convenient for + # binary data. + writer.write(["mess-1", "mess-2"]) # Send multiple messages with one call: + # this reduces overhead on internal SDK processes, + # makes sense with a large message stream. + # [END topic_write] + try: # [BEGIN topic_write_ack] - for message in ["buffer-one", "buffer-two"]: - writer.write(message) - writer.flush(timeout=30) - writer.write_with_ack(["ack-batch-one", "ack-batch-two"], timeout=30) - writer.write_with_ack("ack-one", timeout=30) + # Put several messages into the internal buffer, then wait + # until all of them are delivered to the server. + for mess in messages: + writer.write(mess) + + writer.flush() + + # You can send several messages and wait for acknowledgment for the entire group. + writer.write_with_ack(["mess-1", "mess-2"]) + + # Waiting when sending each message — this method will return a result only after receiving + # acknowledgment from the server. + # This is the slowest message sending option, use it only if this mode + # is really needed. + writer.write_with_ack("message") # [END topic_write_ack] # [BEGIN topic_write_metadata] - message = ydb.TopicWriterMessage(data="metadata", metadata_items={"meta-key": "meta-value"}) + message = ydb.TopicWriterMessage(data="message-data", metadata_items={"meta-key": "meta-value"}) writer.write(message) # [END topic_write_metadata] - + finally: + writer.close() # [BEGIN topic_write_manual] - with driver.topic_client.writer(topic_path, partition_id=0, auto_seqno=False, auto_created_at=False) as writer: - writer.write( - [ - ydb.TopicWriterMessage( - "manual-one", seqno=123, created_at=datetime.datetime.now(datetime.timezone.utc) - ), - ydb.TopicWriterMessage( - "manual-two", seqno=124, created_at=datetime.datetime.now(datetime.timezone.utc) - ), - ] - ) + # Full form, used when, in addition to the message content, you need to manually set its properties. + writer = driver.topic_client.writer(topic=topic_path, auto_seqno=False, auto_created_at=False) + + writer.write(ydb.TopicWriterMessage("asd", seqno=123, created_at=datetime.datetime.now())) + writer.write(ydb.TopicWriterMessage(bytes([1, 2, 3]), seqno=124, created_at=datetime.datetime.now())) + + # In the full form, you can also send multiple messages in a single function call. + # This makes sense with a large stream of outgoing messages — to reduce + # overhead on internal SDK calls. + writer.write( + [ + ydb.TopicWriterMessage("asd", seqno=123, created_at=datetime.datetime.now()), + ydb.TopicWriterMessage(bytes([1, 2, 3]), seqno=124, created_at=datetime.datetime.now()), + ] + ) # [END topic_write_manual] + writer.close() # [BEGIN topic_codec] - with driver.topic_client.writer(topic_path, partition_id=0, codec=ydb.TopicCodec.GZIP) as writer: - writer.write_with_ack("compressed", timeout=30) + writer = driver.topic_client.writer( + topic_path, + codec=ydb.TopicCodec.GZIP, + ) # [END topic_codec] + with writer: + writer.write_with_ack("compressed") def consume(driver, topic_path, consumer_name): @@ -119,59 +198,74 @@ def consume(driver, topic_path, consumer_name): reader = driver.topic_client.reader(topic=topic_path, consumer=consumer_name) # [END topic_start_reader] received = [] - with reader: - if consumer_name == "one": - # [BEGIN topic_read_one] - while len(received) < len(EXPECTED): - message = reader.receive_message(timeout=30) - received.append(check_message(message)) - # [END topic_read_one] - elif consumer_name == "batch": - # [BEGIN topic_read_batch] - while len(received) < len(EXPECTED): - batch = reader.receive_batch(timeout=30) - received.extend(check_message(message) for message in batch.messages) - # [END topic_read_batch] - elif consumer_name == "commit_one": - # [BEGIN topic_read_commit] - while len(received) < len(EXPECTED): - message = reader.receive_message(timeout=30) - received.append(check_message(message)) - reader.commit(message) - # [END topic_read_commit] - elif consumer_name == "commit_batch": - # [BEGIN topic_read_batch_commit] - while len(received) < len(EXPECTED): - batch = reader.receive_batch(timeout=30) - received.extend(check_message(message) for message in batch.messages) - reader.commit(batch) - # [END topic_read_batch_commit] - else: - # [BEGIN topic_soft_stop] - while len(received) < len(EXPECTED): - batch = reader.receive_batch(timeout=30) - received.extend(check_message(message) for message in batch.messages) - reader.commit(batch) - # [END topic_soft_stop] - assert collections.Counter(received) == collections.Counter(EXPECTED) + def process(value): + messages = value.messages if hasattr(value, "messages") else [value] + received.extend(check_message(message) for message in messages) -def check_message(message): - assert message is not None - assert message.data in EXPECTED - if message.data == b"metadata": - # [BEGIN topic_read_metadata] - for key, value in message.metadata_items.items(): - print(key, value) - # [END topic_read_metadata] - assert message.metadata_items["meta-key"] == b"meta-value" - return message.data + with reader: + reader = FiniteReader(reader, received) + try: + if consumer_name == "one": + # [BEGIN topic_read_one] + while True: + message = reader.receive_message() + process(message) + # [END topic_read_one] + elif consumer_name == "batch": + # [BEGIN topic_read_batch] + while True: + batch = reader.receive_batch() + process(batch) + # [END topic_read_batch] + elif consumer_name == "commit_one": + # [BEGIN topic_read_commit] + while True: + message = reader.receive_message() + process(message) + reader.commit(message) + # [END topic_read_commit] + elif consumer_name == "commit_batch": + # [BEGIN topic_read_batch_commit] + while True: + batch = reader.receive_batch() + process(batch) + reader.commit(batch) + # [END topic_read_batch_commit] + elif consumer_name == "soft": + # [BEGIN topic_soft_stop] + while True: + batch = reader.receive_batch() + process(batch) + reader.commit(batch) + # [END topic_soft_stop] + except ScenarioComplete: + pass + assert collections.Counter(received) == collections.Counter(EXPECTED) + + +def metadata(driver, topic_path): + driver.topic_client.create_topic(topic_path, consumers=["metadata"]) + try: + with driver.topic_client.writer(topic_path) as writer: + writer.write_with_ack(ydb.TopicWriterMessage("message-data", metadata_items={"meta-key": "meta-value"})) + with driver.topic_client.reader(topic_path, "metadata") as reader: + # [BEGIN topic_read_metadata] + message = reader.receive_message() + for meta_key, meta_value in message.metadata_items.items(): + print(f"{meta_key}: {meta_value}") + # [END topic_read_metadata] + assert message.metadata_items["meta-key"] == b"meta-value" + finally: + driver.topic_client.drop_topic(topic_path) # [BEGIN topic_no_consumer_handler] -class StartAtBeginning(ydb.TopicReaderEvents.EventHandler): - def on_partition_get_start_offset(self, event): - return ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse(start_offset=0) +class CustomEventHandler(ydb.TopicReaderEvents.EventHandler): + def on_partition_get_start_offset(self, event: ydb.TopicReaderEvents.OnPartitionGetStartOffsetRequest): + return ydb.TopicReaderEvents.OnPartitionGetStartOffsetResponse( + start_offset=0, + ) # [END topic_no_consumer_handler] @@ -179,90 +273,108 @@ def on_partition_get_start_offset(self, event): def without_consumer(driver, topic_path): # [BEGIN topic_no_consumer] - with driver.topic_client.reader( - topic=ydb.TopicReaderSelector(path=topic_path, partitions=[0]), + reader = driver.topic_client.reader( + topic=ydb.TopicReaderSelector( + path=topic_path, + partitions=[0, 1, 2], + ), consumer=None, - event_handler=StartAtBeginning(), - ) as reader: + event_handler=CustomEventHandler(), + ) + # [END topic_no_consumer] + with reader: message = reader.receive_message(timeout=30) check_message(message) - # [END topic_no_consumer] def expired_batch(driver, topic_path): - reader = driver.topic_client.reader(topic=topic_path, consumer="hard") + reader = driver.topic_client.reader(topic_path, "hard") + process = check_message try: - batch = reader.receive_batch(max_messages=1, timeout=30) - assert process_batch(batch) + # [BEGIN topic_hard_stop] + def process_batch(batch): + for message in batch.messages: + if not batch.alive: + return False + process(message) + return True + + batch = reader.receive_batch() + if process_batch(batch): + reader.commit(batch) + # [END topic_hard_stop] finally: - reader.close(flush=False, timeout=30) + reader.close(flush=False) assert not process_batch(batch) -# [BEGIN topic_hard_stop] -def process_batch(batch): - for message in batch.messages: - if not batch.alive: - return False - check_message(message) - return True - - -# [END topic_hard_stop] - - def commit_outside(driver, topic_path): - with driver.topic_client.reader(topic_path, "outside") as reader: + consumer_name = "outside" + with driver.topic_client.reader(topic_path, consumer_name) as reader: message = reader.receive_message(timeout=30) check_message(message) + partition_id = message.partition_id + offset = message.offset + 1 # [BEGIN topic_commit_outside] driver.topic_client.commit_offset( topic_path, - "outside", - 0, - message.offset + 1, - reader.read_session_id, + consumer_name, + partition_id, + offset, + reader.read_session_id, # опционально: не прерывает активную сессию чтения ) # [END topic_commit_outside] -def transactions(driver, topic_path): - transaction_topic = topic_path + "_tx" - driver.topic_client.create_topic(transaction_topic, consumers=["transaction"]) +def transactions(driver, topic): + consumer = "transaction" + message_count = 3 + driver.topic_client.create_topic(topic, consumers=[consumer, "verify"]) try: # [BEGIN topic_write_tx] - with ydb.QuerySessionPool(driver) as pool: + with ydb.QuerySessionPool(driver) as session_pool: - def write_in_transaction(tx): - writer = driver.topic_client.tx_writer(tx, transaction_topic) - for result_set in tx.execute("SELECT 'transaction-message' AS payload"): - payload = result_set.rows[0]["payload"] - writer.write(ydb.TopicWriterMessage(payload)) + def callee(tx: ydb.QueryTxContext): + tx_writer: ydb.TopicTxWriter = driver.topic_client.tx_writer(tx, topic) - pool.retry_tx_sync(write_in_transaction) - # [END topic_write_tx] + for i in range(message_count): + result_stream = tx.execute(query=f"select {i} as res;") + for result_set in result_stream: + message = str(result_set.rows[0]["res"]) + tx_writer.write(ydb.TopicWriterMessage(message)) + print(f"Message {message} was written with tx.") + session_pool.retry_tx_sync(callee) + # [END topic_write_tx] + with driver.topic_client.reader(topic, "verify") as verifier: + payloads = [] + while len(payloads) < message_count: + batch = verifier.receive_batch(timeout=30) + payloads.extend(message.data for message in batch.messages) + assert payloads == [b"0", b"1", b"2"] # [BEGIN topic_read_tx] - with driver.topic_client.reader(transaction_topic, "transaction") as reader: - with ydb.QuerySessionPool(driver) as pool: + with driver.topic_client.reader(topic, consumer) as reader: + with ydb.QuerySessionPool(driver) as session_pool: + for _ in range(message_count): - def read_in_transaction(tx): - batch = reader.receive_batch_with_tx(tx, max_messages=1, timeout=30) - assert batch.messages[0].data == b"transaction-message" + def callee(tx: ydb.QueryTxContext): + batch = reader.receive_batch_with_tx(tx, max_messages=1) + print(f"Message {batch.messages[0].data.decode()} was read with tx.") - pool.retry_tx_sync(read_in_transaction) + session_pool.retry_tx_sync(callee) # [END topic_read_tx] finally: - driver.topic_client.drop_topic(transaction_topic) + driver.topic_client.drop_topic(topic) -def autoscaling(driver, topic_path): +def autoscaling(driver, topic): + consumer = "auto" # [BEGIN topic_autoscale_create] driver.topic_client.create_topic( - topic_path, - consumers=["auto"], - min_active_partitions=1, - max_active_partitions=4, + topic, + consumers=[consumer], + min_active_partitions=10, + max_active_partitions=100, auto_partitioning_settings=ydb.TopicAutoPartitioningSettings( strategy=ydb.TopicAutoPartitioningStrategy.SCALE_UP, up_utilization_percent=80, @@ -271,6 +383,7 @@ def autoscaling(driver, topic_path): ), ) # [END topic_autoscale_create] + topic_path = topic try: # [BEGIN topic_autoscale_alter] driver.topic_client.alter_topic( @@ -283,20 +396,30 @@ def autoscaling(driver, topic_path): ), ) # [END topic_autoscale_alter] - with driver.topic_client.writer(topic_path) as writer: - writer.write_with_ack("auto", timeout=30) - # [BEGIN topic_autoscale_reader] - for full_support in [True, False]: - with driver.topic_client.reader( - topic_path, - "auto", - auto_partitioning_support=full_support, - ) as reader: - message = reader.receive_message(timeout=30) - assert message.data == b"auto" - # [END topic_autoscale_reader] + with driver.topic_client.writer(topic) as writer: + writer.write_with_ack("auto") + # [BEGIN topic_autoscale_reader_full] + reader = driver.topic_client.reader( + topic, + consumer, + auto_partitioning_support=True, # Full support is enabled + ) + # [END topic_autoscale_reader_full] + with reader: + message = reader.receive_message(timeout=30) + assert message.data == b"auto" + # [BEGIN topic_autoscale_reader_compat] + reader = driver.topic_client.reader( + topic, + consumer, + auto_partitioning_support=False, # Compatibility mode is enabled + ) + # [END topic_autoscale_reader_compat] + with reader: + message = reader.receive_message(timeout=30) + assert message.data == b"auto" finally: - driver.topic_client.drop_topic(topic_path) + driver.topic_client.drop_topic(topic) if __name__ == "__main__": From 99d3b839da68124cec1b874a3df98b76302917bd Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Thu, 8 Oct 2026 14:22:05 +0300 Subject: [PATCH 4/7] Keep manual sequence numbers increasing --- examples/ydb_tech/topic/sync_example.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/examples/ydb_tech/topic/sync_example.py b/examples/ydb_tech/topic/sync_example.py index 528875b1..334ffc32 100644 --- a/examples/ydb_tech/topic/sync_example.py +++ b/examples/ydb_tech/topic/sync_example.py @@ -22,6 +22,8 @@ b"asd", bytes([1, 2, 3]), b"compressed", + b"asd", + bytes([1, 2, 3]), ] @@ -177,8 +179,8 @@ def seed(driver, topic_path): # overhead on internal SDK calls. writer.write( [ - ydb.TopicWriterMessage("asd", seqno=123, created_at=datetime.datetime.now()), - ydb.TopicWriterMessage(bytes([1, 2, 3]), seqno=124, created_at=datetime.datetime.now()), + ydb.TopicWriterMessage("asd", seqno=125, created_at=datetime.datetime.now()), + ydb.TopicWriterMessage(bytes([1, 2, 3]), seqno=126, created_at=datetime.datetime.now()), ] ) # [END topic_write_manual] From 53f693c25957e22ef2f0c2b600c2af3027b36894 Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Thu, 8 Oct 2026 14:41:29 +0300 Subject: [PATCH 5/7] Diagnose bounded topic example shutdown in CI --- examples/ydb_tech/topic/async_example.py | 4 ++++ examples/ydb_tech/topic/sync_example.py | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/examples/ydb_tech/topic/async_example.py b/examples/ydb_tech/topic/async_example.py index 61a3a19c..349931b3 100644 --- a/examples/ydb_tech/topic/async_example.py +++ b/examples/ydb_tech/topic/async_example.py @@ -2,6 +2,7 @@ import asyncio import collections +import faulthandler import datetime import os import uuid @@ -339,6 +340,7 @@ async def callee(tx: ydb.aio.QueryTxContext): async def autoscaling(driver, topic): consumer = "auto" + print("Creating autoscaling topic", flush=True) # [BEGIN topic_autoscale_create] await driver.topic_client.create_topic( topic, @@ -353,6 +355,7 @@ async def autoscaling(driver, topic): ), ) # [END topic_autoscale_create] + print("Autoscaling topic created", flush=True) topic_path = topic try: await driver.topic_client.alter_topic( @@ -387,4 +390,5 @@ async def autoscaling(driver, topic): if __name__ == "__main__": + faulthandler.dump_traceback_later(45, repeat=True) asyncio.run(run()) diff --git a/examples/ydb_tech/topic/sync_example.py b/examples/ydb_tech/topic/sync_example.py index 334ffc32..9fdf394f 100644 --- a/examples/ydb_tech/topic/sync_example.py +++ b/examples/ydb_tech/topic/sync_example.py @@ -1,6 +1,7 @@ """Executable source of the synchronous topic snippets on ydb.tech.""" import collections +import faulthandler import datetime import os import uuid @@ -371,6 +372,7 @@ def callee(tx: ydb.QueryTxContext): def autoscaling(driver, topic): consumer = "auto" + print("Creating autoscaling topic", flush=True) # [BEGIN topic_autoscale_create] driver.topic_client.create_topic( topic, @@ -385,6 +387,7 @@ def autoscaling(driver, topic): ), ) # [END topic_autoscale_create] + print("Autoscaling topic created", flush=True) topic_path = topic try: # [BEGIN topic_autoscale_alter] @@ -425,4 +428,5 @@ def autoscaling(driver, topic): if __name__ == "__main__": + faulthandler.dump_traceback_later(45, repeat=True) run() From 1f866587ca51b0ceaf7c07e5a6d216d36c6b1b83 Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Thu, 8 Oct 2026 14:59:08 +0300 Subject: [PATCH 6/7] Exercise initialized Python clients before cleanup --- examples/ydb_tech/topic/async_example.py | 12 ++++++++++-- examples/ydb_tech/topic/sync_example.py | 12 ++++++++++-- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/examples/ydb_tech/topic/async_example.py b/examples/ydb_tech/topic/async_example.py index 349931b3..1a57953a 100644 --- a/examples/ydb_tech/topic/async_example.py +++ b/examples/ydb_tech/topic/async_example.py @@ -72,8 +72,11 @@ async def initialize(topic_path, consumer_name): writer = driver.topic_client.writer(topic_path) reader = driver.topic_client.reader(topic=topic_path, consumer=consumer_name) # [END topic_init] + await writer.write_with_ack("initialization") + message = await asyncio.wait_for(reader.receive_message(), 30) + assert message.data == b"initialization" await writer.close() - await reader.close() + await reader.close(flush=False) async def run(): @@ -93,7 +96,12 @@ async def run(): # [END topic_create] try: await driver.topic_client.alter_topic(topic_path, add_consumers=consumers) - await initialize(topic_path, "init") + initialization_topic = topic_path + "_init" + await driver.topic_client.create_topic(initialization_topic, consumers=["init"]) + try: + await initialize(initialization_topic, "init") + finally: + await driver.topic_client.drop_topic(initialization_topic) # [BEGIN topic_alter] await driver.topic_client.alter_topic( topic_path, diff --git a/examples/ydb_tech/topic/sync_example.py b/examples/ydb_tech/topic/sync_example.py index 9fdf394f..2fdcb978 100644 --- a/examples/ydb_tech/topic/sync_example.py +++ b/examples/ydb_tech/topic/sync_example.py @@ -73,8 +73,11 @@ def initialize(topic_path, consumer_name): writer = driver.topic_client.writer(topic_path) reader = driver.topic_client.reader(topic=topic_path, consumer=consumer_name) # [END topic_init] + writer.write_with_ack("initialization") + message = reader.receive_message(timeout=30) + assert message.data == b"initialization" writer.close() - reader.close() + reader.close(flush=False) driver.stop() @@ -95,7 +98,12 @@ def run(): # [END topic_create] try: driver.topic_client.alter_topic(topic_path, add_consumers=consumers) - initialize(topic_path, "init") + initialization_topic = topic_path + "_init" + driver.topic_client.create_topic(initialization_topic, consumers=["init"]) + try: + initialize(initialization_topic, "init") + finally: + driver.topic_client.drop_topic(initialization_topic) # [BEGIN topic_alter] driver.topic_client.alter_topic( topic_path, From 0409ca98c5ee473919298ec09a2c4ec0d5e21a43 Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Thu, 8 Oct 2026 15:13:11 +0300 Subject: [PATCH 7/7] Remove temporary topic example diagnostics --- examples/ydb_tech/topic/async_example.py | 2 -- examples/ydb_tech/topic/sync_example.py | 2 -- 2 files changed, 4 deletions(-) diff --git a/examples/ydb_tech/topic/async_example.py b/examples/ydb_tech/topic/async_example.py index 1a57953a..d92704df 100644 --- a/examples/ydb_tech/topic/async_example.py +++ b/examples/ydb_tech/topic/async_example.py @@ -2,7 +2,6 @@ import asyncio import collections -import faulthandler import datetime import os import uuid @@ -398,5 +397,4 @@ async def autoscaling(driver, topic): if __name__ == "__main__": - faulthandler.dump_traceback_later(45, repeat=True) asyncio.run(run()) diff --git a/examples/ydb_tech/topic/sync_example.py b/examples/ydb_tech/topic/sync_example.py index 2fdcb978..88755e79 100644 --- a/examples/ydb_tech/topic/sync_example.py +++ b/examples/ydb_tech/topic/sync_example.py @@ -1,7 +1,6 @@ """Executable source of the synchronous topic snippets on ydb.tech.""" import collections -import faulthandler import datetime import os import uuid @@ -436,5 +435,4 @@ def autoscaling(driver, topic): if __name__ == "__main__": - faulthandler.dump_traceback_later(45, repeat=True) run()