*kgo.Record pooling support
#1
Merged
Conversation
This file contains 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
ortuman
force-pushed
the
ortuman/reduce-kgo-record-alloc
branch
2 times, most recently
from
1ab7084 to
0fcbe13
Compare
ortuman
force-pushed
the
ortuman/reduce-kgo-record-alloc
branch
from
0fcbe13 to
fc5a7a3
Compare
Signed-off-by: Miguel Ángel Ortuño <ortuman@gmail.com>
ortuman
force-pushed
the
ortuman/reduce-kgo-record-alloc
branch
from
a85f7a4 to
7dbbd49
Compare
ortuman
commented
Sep 23, 2024
| // | ||
| // This option is particularly useful for use cases where the volume of generated records is very high, | ||
| // as it can negatively impact performance due to the extra GC overhead. | ||
| func EnableRecordsPool() ConsumerOpt { |
There was a problem hiding this comment.
Truth is, I'm not 100% convinced about adding this option. For simplicity, we could always fetch the records directly from the pool, since if the user doesn't explicitly invoke the Reuse method, these will essentially always end up being allocated from the heap.
replay
reviewed
Sep 23, 2024
replay
reviewed
Sep 23, 2024
ortuman
added a commit
that referenced
this pull request
Oct 3, 2024
* fetching: export utilities for decompressing and parsing partition retch responses ### Background In grafana/mimir we are working towards making fetch requests ourselves. The primary reason behind that is that individual requests to the kafka backend are slow, so doing them sequentially per partition becomes the bottleneck in our application. So we want to fetch records in parallel to speed up the consumption. One difficulty I met when issuing `FetchRequest`s ourselves is that parsing the response is non-trivial. That's why I'm proposing to export these functions for downstream projects to use. Alternatively, I can also try contributing the concurrent fetching logic. But I believe that is much more nuanced and with more tradeoffs around fetched bytes and latency. So I wasn't sure whether it's a good fit for a general purpose library. I'm open to discuss this further. ### What this PR does Moves `(*kgo.cursorOffsetNext).processRespPartition` from being a method to being a standalone function - `kgo.processRespPartition`. There were also little changes necessary to make the interface suitable for public use (like removing the `*broker` parameter). ### Side effects To minimize the necessary changes and the API surface of the package I opted to use a single global decompressor for all messages. Previously, there would be one decompressor per client and that decompressor would be passed down to `(*cursorOffsetNext).processRespPartition`. My understanding is that using different pooled readers (lz4, zst, gzip) shouldn't have a negative impact on performance because usage patterns do not affect the behaviour of the reader (for example, a consistent size of decompressed data doesn't make the reader more or less efficient). I have not thoroughly verified or tested this - Let me know if you think that's important. An alternative to this is to also export the `decompressor` along with `newDecompressor()` and the auxiliary types for decompression. * Restore multiline processV0OuterMessage * `*kgo.Records` pooling support Signed-off-by: Miguel Ángel Ortuño <ortuman@gmail.com> * Merge pull request #1 from grafana/ortuman/reduce-kgo-record-alloc `*kgo.Record` pooling support * fetching: export utilities for decompressing and parsing partition retch responses * Merge pull request #4 from dimitarvdimitrov/dimitar/grafana-master-with-export-partition-parsing-utils fetching: export utilities for decompressing and parsing partition fetch responses * Merge pull request #3 from ortuman/reduce-decompression-buffer-allocations Signed-off-by: Miguel Ángel Ortuño <ortuman@gmail.com> --------- Signed-off-by: Miguel Ángel Ortuño <ortuman@gmail.com> Co-authored-by: Dimitar Dimitrov <dimitar.dimitrov@grafana.com>
4 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
Allow reusing
*kgo.Recordobjects returned fromPollFetches/PollRecordsmethods.