Question - "dynamic subscriptions" for direct consumer

[superdodd]

Hello - I have a question about the proper usage pattern of the kgo client in a situation where a given client may need to start and stop consumption of (many different) individual partitions over the course of its lifetime. In brief, we have a way of externally assigning partitions for consumption (either as active or standby) - in either case the partition must be consumed, but the contents may be treated differently. Over time, active or standby partitions may be assigned or unassigned to a given instance as load characteristics change.

Is there a way - easy, or even just correct - of handling this with a single long-lived kgo.Client? It seems like it should be possible to add (and remove) individual partition subscriptions in a running client without affecting the pipeline of fetched/buffered records for other partitions, but I can't see how to do it given the public API of the client. Our current approach is to close and recreate the client entirely when any partition subscriptions change, but this seems a little harsh in order to, for example, add the 100th partition subscription (because it could involve re-fetching any buffered data for the 99 other partition subscriptions that didn't change).

In particular, I guess I'm looking for something like AddConsumePartitions and PurgePartitionsFromClient that work at a partition level rather than a whole-topic level.

[twmb]

Sorry for the delay. Are you using a consumer group or no? AddConsumePartitions would only be relevant for a direct consumer. PurgePartitionsFromClient is not possible because the client itself internally knows of a partition because it knows of a topic, you can't purge a partition because the client must still know of the topic itself (and thus the partition).

If you are using a direct consumer: can you use a combination of AddFetchTopics, PauseFetchPartitions, and ResumeFetchPartitions, PurgeTopicsFromClient to:

• If you need to consume a partition on a new topic, add the topic then pause all unrelated partitions
• Resume partitions you need to consume
• Purge a topic if you decide you do not need to consume any partitions in it anymore
  Note that there will be a few stray messages after adding a topic and pausing (and clearing what's buffered). If you need to exactly consume specific offsets and avoid consuming more / resume exactly at where you wanted to leave off, you'll also have to use SetOffsets, but that gets a little tricky (although it's easier in a direct consumer).

I think AddConsumePartitions is feasible from a feature request standpoint, to avoid the need to add a full topic then pause everything unrelated. PauseFetchPartitions suffices for your PurgePartitionsFromClient request, and then a full purge once you do not want to consume any partitions in the topic anymore.

None of this applies for a group consumer, because as a group consumer you do not actually own which partitions you are consuming.

Let me know if the above add / pause / resume idea works. I'll consider AddConsumePartitions for a feature request for an eventual v1.14 but there still isn't much in the pipeline for that, and if the above idea works, it's good for the meantime.