Add unit tests for asyncio producer + consumer #2036
Merged
Conversation
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
| 🎉 All Contributor License Agreements have been signed. Ready to merge. |
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
fangnx changed the title This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
MSeal approved these changes Sep 2, 2025
There was a problem hiding this comment.
Over looks good, I would change the file naming convention to test_aioconsumer.py and test_aioproducer.py OR we should do a follow-up PR to change the test files to not have capital letters for classes past the test_ prefix to follow standard Python convention.
tests/test_AIOConsumer.py Outdated
| | ||
| @pytest.mark.asyncio | ||
| async def test_constructor_behavior(self, mock_consumer, mock_common, basic_config): | ||
| """Test constructor creates consumer with correct configuration.""" |
There was a problem hiding this comment.
Not quite sure what we're testing for -- that _max_workers on executor isn't overwritten by consumer max_workers?
There was a problem hiding this comment.
Yes, it's to cover the logic that we only set max_workers when using the default executor: https://github.com/confluentinc/confluent-kafka-python/blob/async/src/confluent_kafka/aio/_AIOConsumer.py#L24-L33. It's a pretty trivial test but it's nice to have it covered
| def sync_produce(*args, **kwargs): | ||
| callback = kwargs.get('on_delivery') | ||
| if callback: | ||
| # Simulate successful delivery - callback would normally come from poll() |
Updated the naming 👍 The repo has many other files not following the convention. I think we can raise a separate PR to clean them up |
This comment has been minimized.
This comment has been minimized.
MSeal approved these changes Sep 3, 2025
|
Analysis Details14 Issues
Coverage and Duplications
Project ID: confluent-kafka-python |
MSeal added a commit that referenced this pull request Oct 2, 2025
* Asyncio Producer and Consumer implementation with same API as the sync ones * Add unit tests for asyncio producer + consumer (#2036) * update * update * fix producer and add consumer uts * refactor * rename * Add Benchmark Framework for ducktape (#2030) * Integrate Schema Registry with ducktape load tests (#2027) * basic sr test * more tests * update * update * lint * Add ducktape benchmark tests for consumer (sync + async) (#2045) * draft * update and cleanup * add perf comments, add batch_size param to consume test, lint fix * linter fix * Fix linter issues in consumer testing code (#2051) * Add remaining functions missing in async producer & consumer (#2050) * Add semaphore block for ducktape tests (#2037) * Add semaphore block for ducktape tests * Increase kafka start timeout * Increase kafka start timeout * Increase kafka start timeout * Add logs to debug pipeline * Start kafka in kraft mode * Fix directory failures * Fix directory failures * Fix directory failures * templatise path * Fix ductape run * Fix kafka broker listner * Fix ducktape version error * Cleanup * Fix bound voilation should fail tests * Now expand bounds for success * Add schema registry instance * Update Schema Registry hostname * Update Schema Registry hostname * Update Schema Registry hostname * Fix for linux CI environment * Address minor feedback * Fix semaphore * Minor fix after rebase * Add more async consumer unit & integration tests (#2052) * basic rebalance test * rebalance tests * refactor and linter fix * feebdack * refactor and cleanup * update * remove jit imports * Add produce batch api to producer (#2047) * Add integration tests for transactions (#2056) * add tests * cleanup and linter fix * remove jit import * refactor * cleanup * minot rlinter * Update AsyncIO producer architecture to improve performance (#2044) * Fix helper function name to avoid ducktape test discovery * Integrate schema registry with producer sync/async performance test + clean up the old SR test (#2063) * Add comprehensive producer benchmark tests with Schema Registry support - Updated message serialization to use comprehensive structure with all protobuf fields - Implemented proper strategy pattern for sync/async serializers - Added Schema Registry authentication configuration - Fixed JSON serialization issues (schema title, async serializer initialization) - Added performance validation with configurable JSON validation - Enhanced producer strategies with comprehensive Avro, JSON, and Protobuf support * remove * remove confusing msg * Minor: Producer close calls flush() (#2066) * Integrate schema registry with consumer sync/async performance test (#2059) * update * remove auth * cleanup and ensure same msg size * more cleanup * Add comprehensive producer benchmark tests with Schema Registry support - Updated message serialization to use comprehensive structure with all protobuf fields - Implemented proper strategy pattern for sync/async serializers - Added Schema Registry authentication configuration - Fixed JSON serialization issues (schema title, async serializer initialization) - Added performance validation with configurable JSON validation - Enhanced producer strategies with comprehensive Avro, JSON, and Protobuf support * update * Group messages by topic partition before passing to produce_batch API (#2069) * Merge master to async (#2068) * Pre release (#2067) * Attempting to add python versioning to read from project toml and setting beta flag * Updated docs to read project toml version as well * Updated to read from c file for now. Updaed docs and fixed bad AI code * NPI-7572: Add content for AsyncIO Python client (#2070) * Updates for AsyncIO and other improvements * Add updates based on asyncio blog * Add SR updates relatd to AsyncIO * Reorganize content, remove redundancy, and improve content * Edits to diagram and other content * Add why to use this client in both readme files * Improve CHANGELOG title * Add release dates to versions in CHANGELOG * Add release dates back to v2.4.0 * Edits based on feedback * AsyncIO: Only clear messages from buffer if executor passed (#2071) * Fix async producer transaction behavior + add transactional produce benchmark test (#2072) * update * linter fix * Fix the async transaction behavior related to flush() (#2073) * fix * linter * more linter fix * linter and add link * Removed very old librdkafka version checks * Resolved admin import conflict issue * Fix test_version unit test (#2079) * Fix broken tests (#2077) * fix tests * fix linter * Removed set operation from test --------- Co-authored-by: Matthew Seal <mseal@confluent.io> * Async fix buffer cleanup (#2078) * Fix buffer cleanup logic * Add tests * fix linter * Remove SR key * Removed incorrect assert * Change ducktape tests to install more dependencies * Fix semaphore for producer ducktape tests + clean up files that should've been removed (#2081) * update * use warning for producer validate * remove unnecessary assert --------- Co-authored-by: Emanuele Sabellico <esabellico@confluent.io> Co-authored-by: Matthew Seal <mseal@confluent.io> Co-authored-by: Kaushik Raina <103954755+k-raina@users.noreply.github.com> Co-authored-by: Matthew Seal <mseal007@gmail.com> Co-authored-by: Steve Bang <sbang@confluent.io>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
2 participants
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.
What
Checklist
References
JIRA: https://confluentinc.atlassian.net/browse/DGS-21925
Test & Review
Open questions / Follow-ups