fix(processor): Create strategy lazily on first message #317
Conversation
When a series of rebalance events occurs, this avoids recreating the strategy from within callbacks entirely (making on_revoke in particular faster). Instead the strategy is created whenever the first message is being processed. There were ideas to create the strategy eagerly before even subscribing to a topic. However, currently create_with_partitions requires knowledge of the currently assigned partitions. We would have to deprecate this argument if we wanted to make this happen.
| @@ -243,7 +222,6 @@ def on_partitions_assigned(partitions: Mapping[Partition, int]) -> None: | |||
| "Partition assignment while processing strategy active" | |||
| ) | |||
| _close_strategy() | |||
| _create_strategy(partitions) | |||
There was a problem hiding this comment.
Should we assert that self.__processing_strategy is None here, or make sure we close if it is not None? What would happen if we get an assign without a prior revocation? I know that doesn't actually happen today but the code right now is designed to work for incremental assignment as well, which might not be the case after this change.
There was a problem hiding this comment.
not sure i follow, we do close it on the line before exactly when the strategy is not None
There was a problem hiding this comment.
fwiw this part of rebalancing was actually broken with incremental assign, as we would create the strategy with the subset of partitions from this callback instead of the full assignment. with this pr we're now actually only calling _create_strategy with consumer.tell() which should always contain the full assignment
| # Recreate the strategy if the consumer still has other partitions | ||
| # assigned and is not closed or errored | ||
| try: | ||
| current_partitions = self.__consumer.tell() | ||
| if len(current_partitions.keys() - set(partitions)): | ||
| active_partitions = { | ||
| partition: offset | ||
| for partition, offset in current_partitions.items() | ||
| if partition not in partitions | ||
| } | ||
| logger.info( | ||
| "Recreating strategy since there are still active partitions: %r", | ||
| active_partitions, | ||
| ) | ||
| _create_strategy(active_partitions) | ||
| except RuntimeError: | ||
| pass | ||
|
|
There was a problem hiding this comment.
Just a note that we don't get a subset of partitions revoked today as we are using eager rebalancing. But that may change in the future. This code was written this way so it also works with incremental rebalancing, even though we never actually recreate the strategy on partitions revoked.
There was a problem hiding this comment.
yes i think we're now still handling that case correctly. this code effectively moved from the callback into _run_once. both times we recreate the strategy with the return value from consumer.tell(). we're still closing and flushing the strategy if any partition had been revoked.
just the condition under which we create the strategy changed. before, we'd recreate the strategy if we still had partitions assigned, now we're creating the strategy if we have none and consumer.poll() returned a message.
When a series of rebalance events occurs, this avoids recreating the
strategy from within callbacks entirely (making on_revoke in particular
faster). Instead the strategy is created whenever the first message is
being processed.
There were ideas to create the strategy eagerly before even subscribing
to a topic. However, currently create_with_partitions requires knowledge
of the currently assigned partitions. We would have to deprecate this
argument if we wanted to make this happen.