Repository navigation
Conversation
fsspec license metadata is not detectable by licensecheck, causing CI failures. Add to ignore list since it's BSD-3-Clause licensed.
Implements issue #102: Base component for external communication via pub/sub message broker infrastructure. New base classes: - MessageDataReader: Abstract base for reading data from message brokers with connection management, reconnection with exponential backoff, retry logic, message acknowledgment, and chunked/buffered reading. - MessageDataWriter: Abstract base for writing data to message brokers with connection management, reconnection, retry logic, and chunked/buffered writing. Concrete implementations: - GCPPubSubDataReader/Writer: Google Cloud PubSub - AWSSQSDataReader/AWSSNSDataWriter: AWS SQS/SNS - KafkaDataReader/Writer: Apache Kafka Also includes: - Message broker exceptions (ConnectionError, TransientError, PermanentError) - Settings for GCP PubSub, AWS, and Kafka - Optional dependencies in pyproject.toml - Proposal document with design rationale - Comprehensive unit tests (72 new tests)
|
Benchmark comparison for |
- Fix ruff lint errors: import sorting, unused imports, S110 noqa comments - Fix ruff format errors in gcp_pubsub_io.py - Fix mypy overlap errors: remove duplicate fields from ArgsDict TypedDicts - Fix mypy multiple values error: use kwargs.setdefault instead of pop - Remove untracked test data files causing lint failures
|
Benchmark comparison for |
Increase connection establishment sleep in _ZMQPipelineConnectorProxy from 0.1s to 0.5s to allow the proxy subprocess's SUB socket subscription to propagate to XPUB before the sender starts publishing (ZMQ slow joiner problem). Also mark the test as flaky with 3 reruns following the existing pattern used elsewhere in the repo. Fixes: test_process_with_components_run[RayProcess-zmq_connector_cls-zmq_pubsub_proxy=True-10-2.0]
|
Benchmark comparison for |
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
- Fix invalid-raise errors in message_reader.py and message_writer.py by initializing last_exception with a non-None default instead of Optional[Exception] - Remove PublisherClient.close() call in gcp_pubsub_io.py (method does not exist on the client); just set reference to None for GC - Update test to match new disconnect behavior
|
Benchmark comparison for |
The design proposal is not tracked in the repo. Added to .gitignore to prevent accidental re-commit.
|
Benchmark comparison for |
Resolve conflict in tests/integration/test_process_with_components_run.py: keep main's native pytest parametrization (ConnectorCase + indirect + ids) replacing pytest-cases, retaining the flaky marker for the Ray + ZMQ proxy case. Regenerate uv.lock for merged pyproject extras. Verified: ruff check, ruff format, ty check all pass; new ZMQ backend/ channel/proxy unit tests (21) and message data tests (72) pass; merged integration file collects cleanly (25 tests).
|
Benchmark comparison for |
Blocking issues: - An empty broker poll no longer ends the stream. `_receive` returning [] means "nothing yet", so `step()` waits and polls again like `WebsocketReader`; only a subclass raising `NoMoreDataException` closes the IO (B1). - A batch is acknowledged only after its last record is consumed, instead of after the first one. Previously a crash mid-batch lost the unread messages (B2). - `KafkaDataReader._ack` commits the offsets following the processed records, per partition, instead of the consumer position, which could mark unprocessed records as done (B3). - GCP reader and writer dispatch every synchronous PubSub call (client construction, pull, acknowledge, publish, future.result, stop, close) to a worker thread, so they no longer block the shared event loop (B4, B5). Design and consolidation: - `MessageDataReader`/`MessageDataWriter` now extend `DataReader`/`DataWriter`, reusing the buffer, binding and batching logic instead of copying it (W14). - Retry, backoff and reconnect live in one place: `plugboard.utils.retry` with a frozen `RetryPolicy` value object shared by both bases (W13, L1, L2). - A failing reconnect no longer escapes the retry loop; it costs an attempt and the broker error is raised once attempts run out (W12). - Row-to-message encoding is shared as `iter_records`/`encode_records`/ `encode_records_bytes`; the old per-row deque indexing was quadratic in buffer size and duplicated across three writers (W16, W19). - Connection access is serialised with an `asyncio.Lock` across ack and reconnect (W15). - Kafka writer submits all records then flushes once, instead of one round trip per record (W18). SQS deletes ack batches of up to 10 with `delete_message_batch` and reports partial failures rather than pretending success (L3). - The GCP publisher is stopped on disconnect, so a reconnect no longer leaks a channel and its threads (W17). - `MessageBrokerPermanentError`/`TransientError` are now actually raised and honoured: each broker maps its SDK's failures onto the hierarchy, and permanent errors skip the retries that previously burnt against a guaranteed failure (W1). Configuration and packaging: - The GCP/AWS/Kafka settings are read as constructor fallbacks via `resolve_argument`, so `GCP_PUBSUB_PROJECT_ID`, `AWS_REGION` and `KAFKA_BOOTSTRAP_SERVERS` work (W2). - `aws-messaging` now uses `aiobotocore` instead of `aioboto3`, which no longer forces the shared botocore/boto3 stack backwards (W5). - Broker SDKs are declared in the `test` dependency group, which also resolves the three `ty` unresolved-import findings (W10 root cause). - The six concrete broker classes are exported from `plugboard.library` (W3). - Removed the misleading broker-specific ArgsDicts that documented attributes they did not declare (W20). Tests: 126 tests, up from 72, all using teardown-aware monkeypatch fixtures instead of global `sys.modules` mocks, asserting broker calls rather than private attribute identity (W10, W11). New coverage for retry exhaustion and the delay cap, empty-batch handling, error-branch mapping, offset commits, reconnect failure, and the acknowledgment timing that previously lost records (W7, W8, W9). Record encoding is parametrised once rather than copied across three modules (L4). Docs: new Message Data usage page (extras, worked example, delivery semantics), broker settings in the configuration page, and the new extras in the README (W4). Test isolation: the S3 file tests pinned their bucket region so they no longer depend on the developer's ambient AWS profile, which the newer botocore resolves over `AWS_REGION`.
The connection delay and the flaky marker on the Ray + ZMQ integration test fix an unrelated ZMQ connector issue; they now move with it to their own change.
|
Benchmark comparison for |
…ure at teardown Closes the two remaining uncovered branches from the error-handling paths the review asked about: a reconnect whose disconnect raises must still connect and let the retry proceed, and a send still in flight when destroy runs must be reported rather than swallowed. plugboard/utils/retry.py is now at 100% branch coverage.
Contributor
Author
Review responseAddressed in
Deferred (W6): real-infrastructure integration tests, pending the AWS account and GCP project plus the Terraform in |
|
Benchmark comparison for |
Kafka is not part of the current testing round: there is no broker available to integration-test against, and whether to run one (managed MSK/Confluent versus a container) is still an open decision. Everything Kafka-specific moves to `feat/kafka-message-data`, which stacks on this branch: - `plugboard/library/kafka_io.py` and `tests/unit/test_kafka_io.py` - the `kafka` extra and `aiokafka` in the test dependency group (dropped from the lockfile, so `async-timeout` goes too) - `_KafkaSettings` and the `Settings.kafka` field - the Kafka exports, doc rows, install extra, settings table row and example The message data base classes, the PubSub and SQS/SNS implementations, the shared retry/backoff helper and the docs structure all stay: none of them depended on Kafka.
chrisk314
added a commit
that referenced
this pull request
Oct 18, 2026
Adds `KafkaDataReader` and `KafkaDataWriter` on the `MessageDataReader` / `MessageDataWriter` bases: - Consumer uses a consumer group with auto-commit disabled; acknowledging a batch commits the offsets following the processed records, highest per partition, so records still buffered remain eligible for redelivery. - Producer submits a batch's records together and flushes once, keeping the client's internal batching instead of one round trip per record. - `bootstrap_servers` falls back to `KAFKA_BOOTSTRAP_SERVERS` and accepts a string or a list; aiokafka failures are mapped onto the broker exception hierarchy (authorization / record-too-large are permanent, connection and rebalance errors are retried). - The `kafka` extra declares aiokafka, and aiokafka is in the test dependency group so the unit tests can import the real error and offset types. 372 lines of tests cover offset commits (including out-of-order records across partitions), empty polls returning no messages, multi-partition flattening, batched sends with a single flush, error mapping, settings resolution and JSON vs raw encoding. Split out of #284, where it landed with the PubSub and SQS/SNS implementations.
|
Benchmark comparison for |
Contributor
Author
Kafka split out (#298)Kafka is no longer part of this PR. Everything Kafka-specific moved to the draft branch
Review items that concerned Kafka code specifically now live in #298 rather than here: B3 (commit processed offsets rather than the consumer position), W18 ( Nothing else moved: the base classes, Why draft: there is no broker to integration-test against, and standing one up is still an open decision (managed MSK/Confluent versus a container such as Redpanda in |
Fix two bugs found by running against real infrastructure: - AWSSQSDataReader._ack used the wrong parameter name (DeleteMessageRequests instead of Entries) for delete_message_batch, causing every ack to fail with a validation error. - AWSSQSDataReader._convert did not unwrap the SNS notification envelope that SQS delivers when a queue is subscribed to an SNS topic. Add _decode_body to transparently extract the nested record. Add gated integration tests (tests/integration/test_message_data_brokers.py) that verify real SNS→SQS and PubSub topic→subscription round-trips. Tests are skipped unless PLUGBOARD_MESSAGE_BROKER_TESTS is set, and the CI workflow enables them in the test-integration job with the Terraform-provisioned credentials.
|
Benchmark comparison for |
Before each test the queue/subscription is checked and, if non-empty, a warning is emitted and it is drained, then the clean state is asserted. After each test it is drained again so a failed run leaves no messages to interfere with the next one. - AWS: check via ApproximateNumberOfMessages, drain via purge + receive-and-delete (SQS is the sink for the SNS fan-out, so anything left on it is stray). - GCP: a single timed pull-and-ack drain both checks and empties the subscription (only pubsub.subscriptions.pull/acknowledge are needed).
|
Benchmark comparison for |
Blocking: - B1: MessageDataReader.step now returns after io.close() (not raise), so an exhausted reader completes instead of being marked FAILED. - B2: GCPPubSubDataReader._receive returns [] on a DeadlineExceeded idle pull instead of treating it as a broker error. - B3: RetryPolicy.run now reconnects at the start of each retry attempt and skips that attempt when the reconnect fails, so the operation never runs on a disconnected client; on exhaustion the last real broker error is raised rather than a synthetic "not connected" error. - B4: Removed unnecessary _connect() calls from pure _convert unit tests so they never reach a real cloud client. - B5: Serialised the test-integration job with a concurrency group and gated PLUGBOARD_MESSAGE_BROKER_TESTS to the 3.12 matrix leg so the shared SQS queue and GCP subscription are not raced by concurrent jobs. Warnings: - W1: MessageDataReader._ack_pending retries acknowledgments through the policy. - W2: gcp_pubsub_io._wrap_error maps NotFound to NoMoreDataException only when the reader is exhausted, else a permanent error. - W3: GCP _decode_payload falls back to a "data" field for non-UTF-8, non-JSON, and JSON scalar/array payloads. - W4: MessageDataReader/WriterArgsDict now inherit the DataReader/DataWriter specs; topic is NotRequired with an empty default. - W5: Moved retry execution onto RetryPolicy.run and dropped the module-level helpers that exceeded the parameter limits. - W6: resolve_argument takes keyword-only names and rejects empty resolutions. - W7: Fixed the Raises sections on _fetch/_save and removed unimplemented MessageBrokerConnectionError claims from _connect. - W8: Export RetryPolicy from plugboard.utils and plugboard.library; removed the internal resolve_argument from the public exports. - W9: Assert the broker is empty right after reading and drain it in finally. - W10: Drive the integration writer through its public Component lifecycle (AsyncioConnector inputs + step + io.close + run) and assert region / subscription path via recorded client calls instead of private attributes. - W11: mock_s3_server now scopes its env vars with a MonkeyPatch context and keeps its temp file in a pytest temp dir. - W12: Added unit tests for the _decode_body / _decode_payload fallbacks and the _ack / _disconnect error-wrapping branches. - W13: Fixed the docs example to use AsyncioConnector-wrapped connectors, async with process, a transform that uses both inputs, and the project env var. - W14: Condensed the message-data guide to user-facing bullets (including the writer chunk_size default) and linked the config table instead of duplicating. Low: - L1: parse_json is keyword-only in the shared encoders and broker constructors. - L2: Extracted _parse_json_or_raw to de-nest _decode_body. - L3: Removed the never-read project/subscription/topic attributes on GCP. - L4: Trimmed the remaining rationale-narrating docstrings. - L5: Shared the backoff_delays fixture in conftest so the writer retry tests run without real sleeps. - L6: _drain_gcp_subscription catches DeadlineExceeded specifically. - L7: Extracted a _log_leftover helper in the integration test. - L8: Fixed the "evironment" typo and the /latest/ docs link in the README. - L9: Removed the section-banner comments from the new test modules. - L10: Deleted the superseded local proposal doc and its .gitignore entry.
|
Benchmark comparison for |
The broker tests must run on every supported Python version, but they share a single SQS queue and GCP subscription, so they cannot run at the same time. - Split the broker tests out of the parallel test-integration matrix into a dedicated test-message-brokers job that runs them for 3.12, 3.13 and 3.14 in turn (a per-version uv loop in one job). A single job per run contributes at most one member to its concurrency group, so it is serialized across runs without ever cancelling a waiting request — the failure mode the previous shared-group matrix leg hit. - Drop the concurrency group and the 3.12-only gate from test-integration, and exclude the broker test file so the matrix legs no longer touch the brokers. - Move the GCP round-trip emptiness assertion out of finally and before the post-test drain, so a leftover unacked message is caught instead of being silently removed (and an assert no longer masks an exception from the try).
|
Benchmark comparison for |
The message broker tests must run on every Python version, but they share a single SQS queue and GCP subscription, so they cannot run in parallel. Instead of forcing them into one serial job, each matrix leg now uses its own dedicated broker lane (a separate SNS+SQS pair and Pub/Sub topic+subscription per Python version, provisioned in plugboard-infra). - Drop the serial test-message-brokers job and the --ignore of the broker test file; the broker round trips run again in the normal parallel test-integration matrix, one leg per version. - Each leg resolves its lane from the new BROKER_LANES_JSON repository variable (keyed by python version) with jq, exporting only its own lane's queue/topic identifiers, so legs never contend on a shared broker. - If the lanes variable is absent (e.g. a fork pull request) the broker tests self-skip; a present-but-missing lane for a version fails loudly.
|
Benchmark comparison for |
This branch has not been deployed
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
1 participant
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.
Summary
Implements #102: feat: Base component for external communication (#102).
Adds
MessageDataReader/MessageDataWriterfor reading and writing records through a pub/sub broker - the broker equivalent ofFileReader/FileWriter, where each message carries one record andfield_namesbecome the component's outputs (reader) or inputs (writer). Both extend the existingDataReader/DataWriterbases, adding what a broker needs and a finite source does not: a long-lived connection, reconnection with backoff, and acknowledgment.Two implementations are included: Google Cloud PubSub and AWS SQS/SNS. Apache Kafka is split out into draft PR #298 - see the note at the bottom for why.
Integration tests against real infrastructure are deliberately deferred - no AWS/GCP accounts or broker resources exist yet. Terraform for the required PubSub topic/subscription and SNS topic/SQS queue lives in
plugboard-infra(terraform/ci-data-message-testing), andplugboard-infra/docs/running-message-data-tests.mddescribes wiring its outputs into this suite. All 104 tests here are mock-based, so no credentials are needed.Changes
Base classes
MessageDataReader/MessageDataWriterextendDataReader/DataWriter(buffer, input binding and batching are inherited, not copied), adding_connect,_disconnect,_receive/_send,_convert, and_ack(reader).plugboard/utils/retry.py, with a frozenRetryPolicyvalue object used by both bases.MessageBrokerConnectionError/TransientError/PermanentError) are raised and honoured: each implementation maps its SDK's failures onto them, and permanent errors skip retries that cannot help.GCP_PUBSUB_PROJECT_ID,AWS_REGION); explicit arguments win.plugboard.library.Implementations
GCPPubSubDataReader/GCPPubSubDataWriter(google-cloud-pubsub) - every synchronous PubSub call is dispatched to a worker thread, so a pull or publish never stalls the shared event loop.AWSSQSDataReader/AWSSNSDataWriter(aiobotocore) - long polling,delete_message_batchfor acks, and the 10-message SQS cap is logged rather than applied silently.Packaging: optional extras
gcp-pubsubandaws-messaging; the broker SDKs are also declared in thetestdependency group, which is what lets the tests import the real client and error types.Docs: new Message Data usage page (extras, worked example, delivery semantics), broker settings in the configuration page, new extras in the README.
Behaviour worth knowing
WebsocketReader. A subclass signals a genuinely exhausted source by raisingNoMoreDataException, which closes the IO as other data readers do.Testing
monkeypatchfixtures rather than globalsys.modulesmocks, asserting broker calls and arguments rather than private attribute identity.utils/retry.py100%,message_reader.py95%,message_writer.py94%,gcp_pubsub_io.py94%,aws_messaging_io.py95% - the remaining misses are abstract-method bodies.ruff check,ruff format --check,tyoverplugboard/,plugboard-schemas/andtests/pass;mkdocs buildis clean.test_process_stop_event[LocalProcess-ZMQConnector-...]), which passes on repeat runs in isolation and is addressed in fix: avoid ZMQ slow-joiner loss on pipeline connect #297.Related
plugboard-infra/terraform/ci-data-message-testing- infrastructure for the deferred real-broker integration tests.