[feature] topic post limits #99
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #99 +/- ##
==========================================
+ Coverage 76.22% 76.50% +0.28%
==========================================
Files 32 35 +3
Lines 4076 4529 +453
==========================================
+ Hits 3107 3465 +358
- Misses 776 834 +58
- Partials 193 230 +37
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
There was a problem hiding this comment.
Pull request overview
This PR implements per-topic and broker size limits discovery to prevent oversize messages from being sent, particularly in transactional produce scenarios where oversized messages would previously clog the system. The feature proactively fetches and enforces Kafka broker and topic-level message size limits, returning early errors for messages that exceed these limits.
Key changes:
- New size cap subsystem that asynchronously discovers broker and per-topic message size limits
- Integration of batch processing support in the pwork worker pool via
ItemsWork - Introduction of
multi.Contextfor managing multiple contexts in long-running background operations
Reviewed changes
Copilot reviewed 17 out of 18 changed files in this pull request and generated 11 comments.
Show a summary per file
| File | Description |
|---|---|
| limits.go | New file implementing size cap discovery and enforcement, including broker-level and topic-level limit fetching and validation |
| produce.go | Integrates prepareToProduce to check message sizes before sending, replacing the previous topic creation approach |
| topics.go | Removes createTopicsForOutgoingEvents function, now handled by prepareToProduce |
| internal/pwork/pwork.go | Adds ItemsWork field for batch processing multiple items together, with context sharing via multi.Context |
| internal/multi/multi.go | New multi-context implementation that remains active as long as any constituent context is active |
| connection.go | Adds size cap subsystem fields and helper methods (getABrokerID, getBrokers, findABroker) |
| eventtest/oversize.go | New test validating size limit enforcement across topics with different limits |
| eventtest/happy_path.go | Shortens consumer group names to avoid length issues |
| eventtest/common.go | Improves test name generation, increases timeouts, adds oversize test to matrix |
| eventtest/dead_letter.go | Adds logging for consumer group name |
| consume.go | Adds clarifying comments for waitgroup increment tracking |
| go.mod, go.sum | Adds github.com/chrismcguire/gobberish dependency for test data generation |
| id_allocator_test.go | Minor formatting improvements (whitespace cleanup) |
| events2/shared_test.go, eventpg/shared_test.go, eventnodb/shared_test.go | Adds logging at test start |
| consumer_group_test.go | Adds logging at test start |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 17 out of 18 changed files in this pull request and generated 9 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
S2BryanW
left a comment
S2BryanW
left a comment
There was a problem hiding this comment.
Lots of complex code that I could not review in detail. General flow seems OK, but check comments.
muir
force-pushed
the
prefix
branch
2 times, most recently
from
92d7a57 to
9b8d70e
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 18 out of 19 changed files in this pull request and generated 2 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
On top of #99 This changes the tracer API from passing in a Logf function to instead passing in a function that returns a Logf function given a context. It also introduces an optional TracerConfig that allows spans to be put into context. Specifically: - `Configure` now takes a `TracerProvider` instead of a `Tracer` - Library now supports: `SetTracerConfig(TracerConfig)` to allow spans to be put into contexts The above change is the only API change that is likely to cause breakage but there are more: Additionally: - `eventmodels.LibraryInterface` has more methods - `eventmodels.HandlerInfo` has more methods - `eventmodels.CanValidateTopics` was removed - `eventpg.ProduceDroppedTxEvents` no longer takes a tracer - `eventpg.SaveEventsInsideTx` no longer takes a tracer and takes an `eventmodels.Producer` instead of a `CanValidateTopics` - `events2.Connection.SaveEventsInsideTx` no longer takes a tracer - `events2.ProduceDroppedTxEvents` no longer takes a tracer - `events2.SaveEventsInsideTx` no longer takes a tracer and takes an `eventmodels.Producer` instead of a `CanValidateTopics` Other changes: - some `eventtest` code moved to `eventtest/eventtestutil` - `internal/multi` was removed - lifetimes are now more carefully tracked so that spans can be closed - `Libary.Shutdown()` was added so that tests could easily wait for the library to completely terminate - Now depends on a new version of the simultaneous library - threadContext() is introduced for getting span contexts for background threads - New debug environment variable: EVENTS_DEBUG_CONSUME_STOP
With this change, per topic and broker size limits are discovered and used to prevent oversize messages from being sent. Previously, when using transactional produce, messages could be queued that couldn't be sent. This would clog things up. Now, such messages are stopped early and and error is returned.
Resolves #83
On top of #105