showuon commented on code in PR #16653:
URL: https://github.com/apache/kafka/pull/16653#discussion_r1688008683
##########
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();
+ return null;
+ });
+ } else if
(REMOTE_LOG_DISABLE_POLICY_DELETE.equals(remoteLogDisablePolicy)) {
+ // update start offset to local log start offset
+ // we use unused "-2" as the flag to set local log
start offset
+ updateRemoteLogStartOffset.accept(tp,
FLAG_TO_SET_LOCAL_LOG_START_OFFSET);
+ }
+
removeRemoteTopicPartitionMetrics(tpId);
+ // we already set it correctly, just apply it.
Review Comment:
Since we'll stop RLMM when "delete" is set, I just use "delete" as an
indication to stop RLMM, without introducing a new field.
--
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]
