fetching: export utilities for decompressing and parsing partition fetch responses

[dimitarvdimitrov]

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 FetchRequests 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.

Note to reviewers

I haven't added explicit tests for this because it's not new code and consumer_direct_test.go already tests it. Happy to add tests if you think they're necessary now that this is exported.

[twmb]

I'm open to this. I'm probably going to look more closely at this PR within the next 3w (I'm only choosing empty weekends at the moment for feature work). I'm not entirely convinced on the options struct, I'll see if I can come up with a different proposal.

[dimitarvdimitrov]

Thanks for taking a look.

I also tried to think about alternatives to the options struct: we can create a FetchParser (name TBD ofc) struct that's constructed with a constructor which takes the same parameters as the options fields. Then ProcessRespPartition is a method on that struct instead of just a standalone function.

[gaffneyc]

We ran into this problem as well today looking to convert a custom Fetch from Sarama to Franz which pulled a single record given topic, partition, and offset. I haven't looked at the PR itself but would love to see a way in the API to parse the responses into records.

[twmb]

This is an excellent change. I tried thinking of an alternative API for a while, but everything I thought of comes back to your solution here. The one change I'd make is to export the decompressor and compressor, and then add the decompressor as a field (rather than as a global). I'm making that change and incorporating your changes here in a local PR that I also hope addresses #823 at the same time.

[twmb]

If you're curious, my current mostly-done WIP is implemented in #904.

[twmb]

Closing in favor of the WIP implementation in #904 (which takes into account configurable decompressors, as well as some other things I'm adding in)

[twmb]

This is released.

[gaffneyc]

Awesome, thank you for all your hard work on this release!