Can alter offsets on RebalanceBeforeAssign. #180
Conversation
This is useful if you don't want the broker to manage group offsets on your behalf, but you rather store the offsets locally (typically in a transaction together with the effects of the processing of that batch of messages). See the documentation of rd_kafka_conf_set_rebalance_cb (https://docs.confluent.io/2.0.0/clients/librdkafka/rdkafka_8h.html#a10db731dc1a295bd9884e4f8cb199311).
|
Thanks, looks good! |
|
@alexbiehl if you follow semver strictly, then I guess yes. But let's wait a bit with bumping versions, as I expect to PR one or two smaller fixes yet, if you don't mind. Thanks for the project by the way! |
|
Probably fixes #68. |
|
Thanks! Interesting, CI fails. I need to find time to look at it... |
| Nothing -> pure (pls, ps) | ||
| Just alt -> do | ||
| pls' <- toNativeTopicPartitionList alt | ||
| pure (pls', alt) |
There was a problem hiding this comment.
Can we add some code to guarantee that all (and only) the partitions from the "original" assignment are passed to rdKafkaAssign?
Currently there is a possibility for me to return Just a subset (or a superset) of the partitions. Which may cause problems.
But if we take a list of altered partitions, add the "missing" ones and remove "extra" ones then it should be fine?
There was a problem hiding this comment.
Hey, sorry for the late reply. How about returning IO (Maybe (TopicPartition -> Maybe PartitionOffset)), where an inner Nothing result would mean "don't change offset". This leaves no questions what can and can't be done.
If this sounds right for you and have the capacity, I don't mind you adding to the PR. Otherwise feel free to defer to me, just expect slight delay. Thanks for looking at this!
There was a problem hiding this comment.
@robinp Hmm... From one side - yes... From the other side, do we know that we can demand a pure function here?
Perhaps IO (Maybe (TopicPartition -> IO (Maybe PartitionOffset))), but it gets clunky...
In my mind the original semantics IO (Maybe [TopicPartition]) would read as "give me all you know and I'll do what I can". I.E. I could always return all the offsets of for all the partitions (for the whole topic) if I wanted, but the internals of the callback would figure out the applicable ones for a given consumer in a group...
For that I would even suggest simplifying it to IO [TopicPartition] and say that "we will use offsets that you provide for the partitions that we actually assign" and that'd be it?
There was a problem hiding this comment.
IO-in-IO is not really needed, since the callback already runs in IO, it can query whatever it needs in IO, create the pure structures, and return the offset function based on those pure structures. That practically says, by the time the callback returns, it made up its mind about the offset decisions it would take.
Returning a [TopicPartition] would be deceiving, since suggests it would be used as-is, but as you say, we would need to make adjustments to it to fulfill the Kafka contract.
If we want to cut down on the typesig a bit, then IO (TopicPartition -> Maybe PartitionOffset) can work, as one can always return const Nothing, but an alias like dontOverrideOffsets can be added as well.
There was a problem hiding this comment.
True, we can compute all the offsets and then construct a pure function and return it... That won't even need the first Maybe in this case, IO (TopicPartition -> Maybe PartitionOffset) would be sufficient...
Or perhaps even IO (TopicPartition -> PartitionOffset) since we can always project the original offset into the new one... But I see how this redundant Maybe can be appealing here...
Well, I myself would be happy with just IO [TopicPartition] + merging, since it would be (to me) the most convenient for usage at the price of one bit of knowledge: the returned list cannot alter the list of partitions that are going to be assigned.
But I am not against IO (TopicPartition -> Maybe PartitionOffset) or IO (TopicPartition -> PartitionOffset), too.
One thing that we could do is to keep setRebalanceCallback signature the way it was before, and introduce setRebalanceCallbackWithOffsets with the new signature.
setRebalanceCallback then could just delegate to the new function with a function that doesn't change offsets.
This will allow us to avoid introducing a breaking change.
Do you agree?
There was a problem hiding this comment.
That transition path for smooth upgrade sounds nice.
Yeah, I like IO (TopicPartition -> Maybe PartitionOffset) because I can just const Nothing without having to do a mental exercise of thinking "Can I always return latest? Or unassigned? Or gather the specific value that is available?". I can trust Nothing to do the right thing, hopefully.
Could drop that Maybe, but then some clear guidance is needed in the function doc which value should be returned. Wondering if there's "one true way", in which case rather implement that + have the Maybe, or if there's value in maintaining freedom about multiple behaviors.
There was a problem hiding this comment.
Let's settle on something like
rebalanceCallbackWithOffsets :: (KafkaConsumer -> RebalanceEvent -> IO (TopicPartition -> Maybe PartitionOffset)) -> Callback
rebalanceCallbackWithOffsets = <new implementation>
rebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> Callback
rebalanceCallback f = <call rebalanceCallbackWithOffsets providing (const Nothing)>I guess that it'll be close enough to the best of both having this feature and not breaking compatibility for consumers who don't care about the offsets.
This is useful if you don't want the broker to manage group offsets on
your behalf, but you rather store the offsets locally (typically in a
transaction together with the effects of the processing of that batch of
messages).
See the documentation of rd_kafka_conf_set_rebalance_cb
(https://docs.confluent.io/2.0.0/clients/librdkafka/rdkafka_8h.html#a10db731dc1a295bd9884e4f8cb199311).