feat: Message data classes - #284
Open
chrisk314 wants to merge 9 commits into
Open
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 |
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).
Plugboard's existing DataReader/DataWriter classes handle chunking, buffering, and field-based IO for file and databasesources, but there was no equivalent for message broker infrastructure. This PR introduces MessageDataReader and MessageDataWriter abstract base classes that provide the same field-based IO pattern for pub/sub messaging — with connection lifecycle management, automatic reconnection with exponential backoff, retry logic, message acknowledgment, and chunked/buffered reads and writes. Three concrete implementations are included: Google Cloud PubSub, AWS SQS/SNS, and Apache Kafka. No cloud infrastructure is provisioned yet; all tests run against mocks and will be iterated against real infrastructure later.
Changes
_convert, _ack methods and built-in retry/reconnection
methods and built-in retry/reconnection