Fix duplicate messages when resuming partitions not paused #4636
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
emasab
changed the title
emasab
force-pushed
the
dev_fix_duplicate_messages_cooperative_assignor
branch
from
b228618 to
94fd544
Compare
pranavrth
requested changes
Mar 22, 2024
src/rdkafka_partition.c
Outdated
| /* If partitions isn't paused, avoid bumping its version, | ||
| * as it'll result in resuming fetches from a stale | ||
| * next_fetch_start */ | ||
| rd_bool_t paused = rd_false; |
There was a problem hiding this comment.
there are 2 variables pause and paused. It feels confusing. Please change the name of this variable to is_already_paused
| @@ -2299,7 +2299,22 @@ rd_kafka_resp_err_t rd_kafka_toppar_op_pause_resume(rd_kafka_toppar_t *rktp, | |||
| int flag, | |||
| rd_kafka_replyq_t replyq) { | |||
| int32_t version; | |||
| rd_kafka_op_t *rko; | |||
| rd_kafka_op_t *rko = rd_kafka_op_new(RD_KAFKA_OP_PAUSE); | |||
There was a problem hiding this comment.
This is checked before the operation is sent to the queue. What about the rd_kafka_toppar_pause_resume in which this op is handled. Is this case handled there?
tests/0145-unnecessary_resume_mock.c
Outdated
| * | ||
| * @param partition_assignment_strategy Assignment strategy to test. | ||
| */ | ||
| static void test_no_duplicate_messages_unnecessary_resume( |
There was a problem hiding this comment.
Name of the test should be better. It can be pause_resume_mock.
tests/0145-unnecessary_resume_mock.c
Outdated
There was a problem hiding this comment.
I was thinking to define a better numbering for mock tests. Integration tests starts with '0', unit test with '8', so we can use mock tests with '4' maybe?
| @@ -511,7 +512,7 @@ struct test tests[] = { | |||
| _TEST(0142_reauthentication, 0, TEST_BRKVER(2, 2, 0, 0)), | |||
| _TEST(0143_exponential_backoff_mock, TEST_F_LOCAL), | |||
| _TEST(0144_idempotence_mock, TEST_F_LOCAL, TEST_BRKVER(0, 11, 0, 0)), | |||
|
|
|||
tests/0145-unnecessary_resume_mock.c
Outdated
| test_msgver_init(&mv, testid); | ||
| test_consumer_poll("consume", rk, testid, -1, 0, msgcnt, &mv); | ||
|
|
||
| TEST_SAY("Unnecessary resume\n"); |
There was a problem hiding this comment.
Is this test failing before the change?
emasab
force-pushed
the
dev_fix_duplicate_messages_cooperative_assignor
branch
from
f11be5a to
58e9166
Compare
cooperative assignor
emasab
force-pushed
the
dev_fix_duplicate_messages_cooperative_assignor
branch
from
62583dd to
20bcfd1
Compare
7 tasks
anchitj
pushed a commit
that referenced
this pull request
Jun 10, 2024
In case of subscription change with a consumer using the cooperative assignor it could resume fetching from a previous position. That could also happen if resuming a partition that wasn't paused. Fixed by ensuring that a resume operation is completely a no-op when the partition isn't paused
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.
Fixed a bug causing duplicate message consumption from a stale fetch start offset in some particular cases.
In case of subscription change with a consumer using the cooperative assignor it could resume fetching from a previous position.
That could also happen if resuming a partition that wasn't paused.
Fixed by ensuring that a resume operation is completely a no-op when the partition isn't paused.
Fixes #4637