Clean buffer after rebalancing for batch queue #3269
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
edenhill
reviewed
Feb 18, 2021
edenhill
reviewed
Feb 25, 2021
edenhill
reviewed
Feb 25, 2021
edenhill
suggested changes
Mar 10, 2021
Issue: Buffer is not cleaned after rebalance if messages are polled using the batch queue method, so the consumer will still get old messages. Solution: when assign happens, a new op event with type RD_KAFKA_OP_BARRIER will be created, a new version is been created at mean time. If the consumer met this event, will clean the buffer by comparing the version of msgs and the new version just created.
edenhill
suggested changes
Mar 22, 2021
edenhill
suggested changes
Apr 1, 2021
src/rdkafka_queue.c
Outdated
| if (rd_kafka_op_version_outdated(rko, version)) { | ||
| /* This also destroys the corresponding rkmessage. */ | ||
| rd_kafka_op_destroy(rko); | ||
| } else { |
There was a problem hiding this comment.
Perhaps as an optimization to avoid a write shuffle: else if (i > valid_count++)
There was a problem hiding this comment.
Hi Magnus, thanks for pointing this out! I think copy should before valid_count + 1, so I change this part to the below logic:
if (rd_kafka_op_version_outdated(rko, version)) {
/* This also destroys the corresponding rkmessage. */
rd_kafka_op_destroy(rko);
} else if ((size_t)i > valid_count) {
rkmessages[valid_count++] = rkmessages[i];
} else {
valid_count++;
}
jliunyu
commented
Apr 7, 2021
edenhill
approved these changes
Apr 13, 2021
|
|
||
|
|
||
| /** | ||
| * @brief Produce 400 messages and consume 500 messages totally by 2 consumers |
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.
Summary: This is bug fix for confluentinc/confluent-kafka-python#1013
Issue: Buffer is not cleaned after rebalance if messages are polled using the batch queue method, so the consumer will still get old messages.
Solution: when assign happens, a new op event with type RD_KAFKA_OP_BARRIER will be created, a new version is been created at mean time. If the consumer met this event, will clean the buffer by comparing the version of msgs and the new version just created.
Test passed in local test:
/ 63.161s] ALL-TESTS: duration 63160.668msTesting:
[0122_buffer_cleaning_after_rebalance/ 62.283s] ================= Test 0122_buffer_cleaning_after_rebalance PASSED =================
[
TEST 20210224223610 (bare, scenario default) SUMMARY
#==================================================================#
| | PASSED | 63.161s |
| 0122_buffer_cleaning_after_rebalance | PASSED | 62.283s |
#==================================================================#
[ / 63.163s] 0 thread(s) in use by librdkafka
[ / 63.163s]
============== ALL TESTS PASSED ==============
./test-runner in bare mode PASSED!