kamalcph commented on code in PR #16653:
URL: https://github.com/apache/kafka/pull/16653#discussion_r1686637497
##########
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:
If we cancel the task for "delete" policy, then after moving the
log-start-offset to LLSO, the remote log segments won't be cleared.
For both the policies, we need the expiration task to be running so that the
remote log segments gets removed.
--
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]
