refactor(log): introduce async close

core/src/main/scala/kafka/log/LocalLog.scala

@@ -233,6 +233,15 @@ class LocalLog(@volatile protected var _dir: File,
     // AutoMQ inject end
   }
 
+  /**
+   * Asynchronously closes the segments of the log.
+   */
+  private[log] def asyncClose(): CompletableFuture[Void] = {
+    CompletableFuture.completedFuture(() -> {
+      close()
+    })
+  }
+
   /**
    * Completely delete this log directory with no delay.
    */

core/src/main/scala/kafka/log/UnifiedLog.scala

@@ -650,6 +650,10 @@ class UnifiedLog(@volatile var logStartOffset: Long,
    * The memory mapped buffer for index files of this log will be left open until the log is deleted.
    */
   override def close(): Unit = {
+    asyncClose().get()
+  }
+
+  def asyncClose(): CompletableFuture[Void] = {
     debug("Closing log")
     lock synchronized {
       logOffsetsListener = LogOffsetsListener.NO_OP_OFFSETS_LISTENER
@@ -662,7 +666,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
         // (the clean shutdown file is written after the logs are all closed).
         producerStateManager.takeSnapshot()
       }
-      localLog.close()
+      localLog.asyncClose()
     }
   }
 

core/src/main/scala/kafka/log/streamaspect/ElasticLog.scala

@@ -450,6 +450,9 @@ class ElasticLog(val metaStream: MetaStream,
         newSegment
     }
 
+    /**
+     * ref. LocalLog#close
+     */
     override private[log] def close(): CompletableFuture[Void] = {
         // already flush in UnifiedLog#close, so it's safe to set cleaned shutdown.
         partitionMeta.setCleanedShutdown(true)
@@ -466,6 +469,15 @@ class ElasticLog(val metaStream: MetaStream,
         closeStreams()
     }
 
+    /**
+     * ref. LocalLog#asyncClose
+     */
+    override private[log] def asyncClose(): CompletableFuture[Void] = {
+        CompletableFuture.completedFuture(() -> {
+            close()
+        })
+    }
+
     /**
      * Directly close all streams of the log.
      */

core/src/main/scala/kafka/log/streamaspect/ElasticUnifiedLog.scala

@@ -185,8 +185,7 @@ class ElasticUnifiedLog(_logStartOffset: Long,
         asyncClose().get()
     }
 
-    // TODO: invoke async close
-    def asyncClose(): CompletableFuture[Void] = {
+    override def asyncClose(): CompletableFuture[Void] = {
         ElasticUnifiedLog.Logs.remove(elasticLog.topicPartition, this)
         val closeFuture = lock synchronized {
             maybeFlushMetadataFile()
@@ -200,7 +199,7 @@ class ElasticUnifiedLog(_logStartOffset: Long,
             }
             // flush all inflight data/index
             flush(true)
-            elasticLog.close()
+            elasticLog.asyncClose()
         }
         elasticLog.segments.clear()
         closeFuture.whenComplete((_, _) => {