showuon commented on code in PR #16653:
URL: https://github.com/apache/kafka/pull/16653#discussion_r1687430791
##########
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:
We can make delete async in
[KAFKA-15313](https://issues.apache.org/jira/browse/KAFKA-15313) for sure. For
this PR, I think we just focus on enabling the remote log disable policy
feature.
--
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]
