BlockBuilder: Refactor the consume cycle logic #8329
Merged
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
codesome
commented
Jun 10, 2024
331ea4b to
712b67d
Compare
Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>
712b67d to
af11016
Compare
narqo
reviewed
Jun 11, 2024
narqo
reviewed
Jun 11, 2024
Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>
Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>
codesome
commented
Jun 12, 2024
| @@ -28,7 +28,7 @@ import ( | |||
| ) | |||
|
|
|||
| type builder interface { | |||
| process(ctx context.Context, rec *kgo.Record, blockMin, blockMax int64, recordProcessedBefore bool) (_ bool, err error) | |||
| process(ctx context.Context, rec *kgo.Record, lastBlockMax, blockMax int64, recordProcessedBefore bool) (_ bool, err error) | |||
There was a problem hiding this comment.
I changed the name because technically it is not a blockMin, rather the max of the last cycle
narqo
reviewed
Jun 12, 2024
narqo
reviewed
Jun 12, 2024
Signed-off-by: Vladimir Varankin <vladimir.varankin@grafana.com>
Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>
codesome
commented
Jun 12, 2024
pkg/blockbuilder/blockbuilder.go
Outdated
Comment on lines
507
to
509
| for i := range req.Topics[0].Partitions { | ||
| if req.Topics[0].Partitions[i].Partition == commitRec.Partition { | ||
| req.Topics[0].Partitions[i].Metadata = &meta |
There was a problem hiding this comment.
Fixed the commit metadata. Debugging through the kgo library I saw that this string was not being overridden. By default kgo sets it to member ID (which is what we were seeing).
See 3e5e03f for the fix
There was a problem hiding this comment.
I have written unit tests for this. I will clean it up and push it as well.
Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>
Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>
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.
@narqo this moves some logic out of
consumePartitionand into thenextConsumeCycle. Main aim of this is to makeconsumePartition()simpler where it just starts consuming from the partition until the given cycle end time. We also pass the block start and end times, and the last seen record as part of arguments.We now do all the heavy lifting in nextConsumeCycle, where we call
consumePartition()in parts when it is lagging. We calculate the timestamps appropriately.