[GitHub] [kafka] showuon commented on a diff in pull request #13947: KAFKA-15130: Delete remote segments when delete a topic

["showuon (via GitHub)" <gi...@apache.org>]

showuon commented on code in PR #13947:
URL: https://github.com/apache/kafka/pull/13947#discussion_r1282913466


##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -526,25 +530,37 @@ class ReplicaManager(val config: KafkaConfig,
   /**
    * Stop the given partitions.
    *
-   * @param partitionsToStop    A map from a topic partition to a boolean indicating
-   *                            whether the partition should be deleted.
+   * @param partitionsToStop A map from a topic partition to a boolean indicating
+   *                         whether the partition should be deleted.
+   * @return A map from partitions to exceptions which occurred.
+   *         If no errors occurred, the map will be empty.
+   */
+  protected def stopPartitions(partitionsToStop: Map[TopicPartition, Boolean]): Map[TopicPartition, Throwable] = {
+    stopPartitions(partitionsToStop.map(e => StopPartition(e._1, e._2)).toSet)

Review Comment:
   nit: could we not use `e` as parameter? It will make people think it's a exception.



##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -343,21 +345,78 @@ public void onLeadershipChange(Set<Partition> partitionsBecomeLeader,
     /**
      * Deletes the internal topic partition info if delete flag is set as true.
      *
-     * @param topicPartition topic partition to be stopped.
+     * @param topicPartitions topic partitions that needs to be stopped.
      * @param delete         flag to indicate whether the given topic partitions to be deleted or not.
      */
-    public void stopPartitions(TopicPartition topicPartition, boolean delete) {
+    public void stopPartitions(Set<TopicPartition> topicPartitions,
+                               boolean delete,
+                               BiConsumer<TopicPartition, Throwable> errorHandler) {
+        LOGGER.debug("Stopping {} partitions, delete: {}", topicPartitions.size(), delete);
+        Set<TopicIdPartition> topicIdPartitions = topicPartitions.stream()
+                .filter(topicIdByPartitionMap::containsKey)
+                .map(tp -> new TopicIdPartition(topicIdByPartitionMap.get(tp), tp))
+                .collect(Collectors.toSet());
+
+        topicIdPartitions.forEach(tpId -> {
+            try {
+                RLMTaskWithFuture task = leaderOrFollowerTasks.remove(tpId);
+                if (task != null) {
+                    LOGGER.info("Cancelling the RLM task for tpId: {}", tpId);
+                    task.cancel();
+                }
+                if (delete) {
+                    LOGGER.info("Deleting the remote log segments task for partition: {}", tpId);
+                    deleteRemoteLogPartition(tpId);

Review Comment:
   Agree to delete all the remote logs async. Delegating the responsibility to RLMM is a solution for sure, but that might need to mention in the KIP or somewhere. 



-- 
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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org