Get ConsumerGroups information from ConsumerGroupRegister and update tracked Lag when assigned consumer stops consuming
#131
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
detro
added
enhancement
detro
changed the title
ConsumerGroups will now come from ConsumerGroupRegisterConsumerGroups information from ConsumerGroupRegister
detro
force-pushed
the
detro/114-consumer_groups_register
branch
from
b94ca68
to
acb44d5
Compare
Before the error was being swallowed.
This is able to provide the groups and the members of each group. Additionally, it holds the groups in memory after they shutdown/disappear, for a configurable `forget_after` `Duration`.
…rGroupsRegister` `forget_after` input
…`ConsumerGroupsRegister`
Meaning, not known to the `ConsumerGroupsRegister` anymore
…onsumerGroupsRegister`
…or the specified amount of time
detro
force-pushed
the
detro/114-consumer_groups_register
branch
from
acb44d5
to
5fa7c82
Compare
This means that the group is gone, so we don't want to remove it from the register just yet.
1. Check if `ConsumerGroupRegister` has changed (using the hash) and, if it did, updating it's internal tracked group lags 2. Check if any known `Lag` is stale, and if it is updating the lag estimation considering the latest tracked produced offsets
detro
changed the title
ConsumerGroups information from ConsumerGroupRegisterConsumerGroups information from ConsumerGroupRegister and update tracked Lag when assigned consumer stops consuming
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.
Enables: #114
Here I introduce the
ConsumerGroupRegister.This is not too different from other
*Register, likeClusterStatusRegister: its focus is on holding the information about the available, activeConsumerGroupsin the cluster.In addition to caching the information about
ConsumerGroups, it enacts the role of "remembering the group for a certain amount of time": this is embodied in the--forget-group-afternew command line parameter. This new register is going to not remove any of theConsumerGroupsthat stop being reported by the cluster, for a fixed amount of time.This is to help with the fact that, in case of an outage, a
Groupcould disappear for various reasons (a crash?): if theGroupwas immediately forgotten, it would make the metric produced by Kommitted disappear.This way the metric will keep being produced, and the internal estimator will use the known previous data points to estimating the growth of latency (that is ultimately the whole point of this service).
Additionally, this change allows for #114 to happen: to calculate lag for clusters hosted on Confluent Cloud, we need to refactor how the
KommittedOffsetsDatais produced, by allowing for an alternativeKommittedOffsetsDataEmitterto be created (e.g.ConfluentCloudOffsetsDataEmitter). The API offered by Committed Offsets require a prior knowledge of the available groups, so the register here will work as a source of truth, for when the API calls are done to get to the offset information of the specific group.The
main.rswill have to then swap the defaultKommittedOffsetsDataEmitterwith theConfluentCloudOffsetsDataEmitter, based on input. But this is for a follow up PR.