KAFKA-15181: fix(storage): wait for consumer to be synced after assigning partitions#14012
KAFKA-15181: fix(storage): wait for consumer to be synced after assigning partitions#14012jeqo wants to merge 6 commits into
Conversation
There was a problem hiding this comment.
Implementation looks good with me, also appreciate the minor code improvements. Only question is whether a test is required for this, and if so is there some prior art for testing waitTillConsumptionCatchesUp.
We are also going to test this implementation internally to verify there are no issues.
showuon
left a comment
showuon
left a comment
There was a problem hiding this comment.
One more comment, I don't see tests for the new added method. Could you add some tests for them?
|
@showuon thanks for your feedback! I have applied most suggestions -- let me know if I'm missing anything. About the test, I followed @mdedetrich advise on piggybacking on existing testing of I tried to add some scenarios to |
showuon
left a comment
showuon
left a comment
There was a problem hiding this comment.
LGTM, just a minor comment.
About the test, I followed @mdedetrich advise on piggybacking on existing testing of waitTillConsumptionCatchesUp and actually removing the waiting from TopicBasedRemoteLogMetadataManagerTest to prove the new behavior when assigning partitions.
Fair enough!
kamalcph
left a comment
kamalcph
left a comment
There was a problem hiding this comment.
@jeqo Thanks for the PR!
- Can we split this patch into two? a) Refactoring the existing code and b) Specific changes that address KAFKA-15181 ticket.
- We are instantiating
AdminClient(one more client) to do listOffset calls in ConsumerManager to ensure that once the user-partitions are assigned, all the messages are caught-up from the respective metadata-partition.
Both Consumer and Admin clients, can do list-offset calls. Instead, of adding one more Admin client, can we move the initialisation logic to RemoteLogMetadataCache.java class by introducing a latch. All the calls to read/write from remote will hit the RemoteLogMetadataCache and the ConsumerTask can mark the partition as initialised once it reads all the messages.
There was a problem hiding this comment.
partition{} -> partition {}
There was a problem hiding this comment.
This is a blocking call. It won't return until the partition is ready/consumption-catches-up. In high throughput cluster, this can anywhere take order of seconds to complete.
Moreover, when you reassign the partitions, the controller will send the LISR request for one partition at a time and the LISR request handling path shouldn't have any blocking call so that the controller can send the next LISR request on time. Otherwise, LISR request gets timed out.
There was a problem hiding this comment.
Good point @kamalcph ! Do you have any suggestion here?
There was a problem hiding this comment.
I am also unclear as to what we should do here, I mean we could return a CompletableFuture<Void> but this would involve changing the interface?
There was a problem hiding this comment.
One approach we can take is to wait for metadata partition to catch up asynchronously. All the calls to read the remote log will hit the RemotePartitionMetadataStore#getRemoteLogMetadataCache, there we can ensure that the partition is initialised via a latch.
The ConsumerTask can keep a marker of the latest offset of metadata-partition when user-partitions are assigned to them and can mark that user-partition as initialised in the cache once it read/re-read messages up-to that point.
We can have a latch with timeout of 2 mins to avoid holding the caller thread for a long amount of time and throw a retry-able error.
There was a problem hiding this comment.
Sounds great! I like the idea that we lazily wait for the consumer to catch up. One comment is why don't we return retriable error immediately, and ask the client to try it again, like when the leader is under election, the client will get NOT_LEADER_OR_FOLLOWER error immediately and retry, instead of waiting in the broker side until leader election completed. WDYT?
There was a problem hiding this comment.
That also works! We have to look around for all the callers:
- Replica fetcher thread will retry the request after the
replica.fetch.backoff.msonce the partition is marked as error/failed. - Consumers may resend the FETCH request as soon as it fails without the request quota.
We can also think about adding one more API in RemoteLogMetadataManager to ensure that the partition is initialised before copying the segments (or) when the replica thread is being initialised.
boolean isInitialized(TopicIdPartition topicIdPartition) throws RemoteStorageException;
There was a problem hiding this comment.
The latching mechanism as mentioned by @kamalcph is what we are currently using internally. @abhijeetk88 is planning to pull those changes to trunk and raise PR this week to handle the scenario.
There was a problem hiding this comment.
Thanks @jeqo for the PR. Let us keep the changes separate as @kamalcph suggested in other comment. We should avoid blocking the calls while registering a partition to RLMM, which is handled using latches in internal branch, left a comment on the same. Let us have this PR with the cleanup and other minor changes.
TBRLMM needs to create admin client at separate instances, and common set of configs already existing are enough -- no need to reuse producer/consumer configs
When partitions are assigned on leadership changes cache should be ready to reply, otherwise there is a risk of reads on cache without partitions assigned, and failing. This can happen on bootstrap where partitions are assigned, then added to partitions pending to load, and then taking some time to load data from partitions. The waiting happening on this test should then be actual part of the logic when assigning partitions to TBRLMM, and give the same guarantees as when writing and then waiting for data to be consumed.
This fixes a race condition where cache is not ready but TBRLMM is marked as initialized anyway, causing replica to fail as remote log metadata is not found.
6 participants
This fixes a race condition where cache is not ready but TBRLMM is marked as initialized anyway, causing replica to fail as remote log metadata is not found.
Committer Checklist (excluded from commit message)