showuon commented on code in PR #16653:
URL: https://github.com/apache/kafka/pull/16653#discussion_r1687371692
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -482,19 +492,29 @@ public void stopPartitions(Set<StopPartition>
stopPartitions,
task.cancel();
return null;
});
- leaderExpirationRLMTasks.computeIfPresent(tpId,
(topicIdPartition, task) -> {
- LOGGER.info("Cancelling the expiration RLM task for
tpId: {}", tpId);
- task.cancel();
- return null;
- });
followerRLMTasks.computeIfPresent(tpId, (topicIdPartition,
task) -> {
LOGGER.info("Cancelling the follower RLM task for
tpId: {}", tpId);
task.cancel();
return null;
});
+ // here, we have to handle retain and delete separately.
+ // If "retain" is set, we should not cancel expiration task
+ if
(!REMOTE_LOG_DISABLE_POLICY_RETAIN.equals(remoteLogDisablePolicy)) {
+ leaderExpirationRLMTasks.computeIfPresent(tpId,
(topicIdPartition, task) -> {
+ LOGGER.info("Cancelling the expiration RLM task
for tpId: {}", tpId);
+ task.cancel();
Review Comment:
No, they will get removed because we explicitly delete all the remote log
segments in the method
[below](https://github.com/apache/kafka/blob/trunk/core/src/main/java/kafka/log/remote/RemoteLogManager.java#L500),
not rely on the `leaderExpirationRLMTasks`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
