feat kafka: added seek functionality for consumer.#948
Closed
melonaerial wants to merge 33 commits into
Closed
Conversation
melonaerial
commented
Jun 13, 2025
fdr400
suggested changes
Jun 16, 2025
fdr400
suggested changes
Jun 20, 2025
melonaerial
changed the title
fdr400
suggested changes
Jul 1, 2025
| if (event_type == kafka::RebalanceEventType::kAssigned) { | ||
| for (const auto& topic_partitions : partitions) { | ||
| UEXPECT_NO_THROW(consumer_scope.SeekToBeginning( | ||
| std::string(topic_partitions.topic), topic_partitions.partition_id, poll_timeout |
Collaborator
There was a problem hiding this comment.
давай тоже utils::zstring_view принимать, а то как-то неудобно выходит))
| kafka::utest::Message{topic, "key", "value", 0, kHeaders}, | ||
| kafka::utest::Message{topic, "key2", "value2", 0, kHeaders}, | ||
| }; | ||
| const std::uint64_t kMessagesToSkip = 1; |
Collaborator
There was a problem hiding this comment.
давай сделаем 10 сообщений и 7 в skip
Comment on lines
+394
to
+395
| static constexpr std::array kExpectedHeaders{kafka::HeaderView{"header-1", "value-1"}}; | ||
| const std::vector<kafka::OwningHeader> kHeaders{kExpectedHeaders.begin(), kExpectedHeaders.end()}; |
Comment on lines
+354
to
+356
| const std::array kMessages{ | ||
| kafka::utest::Message{topic, "key", "value", 0, kHeaders}, | ||
| kafka::utest::Message{topic, "key2", "value2", 0, kHeaders}, |
Collaborator
There was a problem hiding this comment.
- давай больше сообщений
- можем сделать два раза Receive? Сначала без сик, прочитываем всё до конца, потом делаем сик, потом заново всё прочитываем
| }); | ||
|
|
||
| const auto* err = rd_kafka_seek_partitions( | ||
| consumer_.GetHandle(), topic_partitions_list.GetHandle(), static_cast<int>(timeout.count()) |
Collaborator
There was a problem hiding this comment.
тут получается 2 * timeout. В этом месте уже должно быть timeout - время на poll
|
|
||
| // Added `rd_kafka_queue_poll` call to activate the assign operation | ||
| // if it is called from RebalanceCallback. | ||
| { EventHolder event{rd_kafka_queue_poll(consumer_.GetQueue(), static_cast<int>(timeout.count()))}; } |
Collaborator
There was a problem hiding this comment.
а если пришёл месседж в эвент? просто скрипнем?
выглядит как-то очень странно, кмк
а если вызывано не из rebalance callback?
Contributor
Author
There was a problem hiding this comment.
Так мы всё равно seek делаем, если мессадж ивент придёт, то нам как бы пофиг на него, мы на другой offset скачем же. Тут самое фиговое если ещё раз ребаланс прилетит, но после успешного assign и revoke такого быть не должно судя по всему.
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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.
Seek,SeekToBeginning,SeekToEnd,SeekToEndfunctions toConsumerScopeclass.RebalanceCallbackfor a consumer which is called after assign/revoke partitions occurs.Note: by creating a PR or an issue you automatically agree to the CLA. See CONTRIBUTING.md. Feel free to remove this note, the agreement holds.