refine

Author: FMX

worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java

@@ -48,7 +48,6 @@ public class PartitionDataWriter implements DeviceObserver {
   protected final DeviceMonitor deviceMonitor;
   private final long splitThreshold;
   private final PartitionSplitMode splitMode;
-  protected boolean deleted = false;
   protected final FlushNotifier notifier = new FlushNotifier();
   // It's only needed when graceful shutdown is enabled
   protected final StorageManager storageManager;
@@ -147,23 +146,10 @@ public long close() {
     return tierWriterProxy.close();
   }
 
-  @FunctionalInterface
-  public interface RunnableWithIOException {
-    void run() throws IOException;
-  }
-
   public boolean isClosed() {
     return tierWriterProxy.isClosed();
   }
 
-  protected synchronized long close(
-      RunnableWithIOException tryClose,
-      RunnableWithIOException streamClose,
-      RunnableWithIOException finalClose)
-      throws IOException {
-    return tierWriterProxy.close();
-  }
-
   public void evict(boolean checkClose) throws IOException {
     // this lock is used to make sure that
     // memory manager won't evict with writer thread concurrently

worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StoragePolicy.scala

@@ -51,7 +51,7 @@ class StoragePolicy(conf: CelebornConf, storageManager: StorageManager, source:
           orderList)
       }
     }
-    logError(s"Create evict file filaed for ${partitionDataWriterContext.getPartitionLocation}")
+    logError(s"Create evict file failed for ${partitionDataWriterContext.getPartitionLocation}")
     null
   }
 
@@ -69,7 +69,7 @@ class StoragePolicy(conf: CelebornConf, storageManager: StorageManager, source:
         "Partition data writer context can not have null partition location")
     }
 
-    def getPartitionMetahandler(fileInfo: FileInfo) = {
+    def getPartitionMetaHandler(fileInfo: FileInfo) = {
       partitionType match {
         case PartitionType.REDUCE =>
           new ReducePartitionMetaHandler(partitionDataWriterContext.isRangeReadFilter, fileInfo)
@@ -98,7 +98,7 @@ class StoragePolicy(conf: CelebornConf, storageManager: StorageManager, source:
               partitionDataWriterContext.getUserIdentifier,
               partitionDataWriterContext.getPartitionType,
               partitionDataWriterContext.isPartitionSplitEnabled)
-            val metaHandler = getPartitionMetahandler(memoryFileInfo)
+            val metaHandler = getPartitionMetaHandler(memoryFileInfo)
             new MemoryTierWriter(
               conf,
               metaHandler,
@@ -120,7 +120,7 @@ class StoragePolicy(conf: CelebornConf, storageManager: StorageManager, source:
               partitionDataWriterContext.getPartitionType,
               partitionDataWriterContext.isPartitionSplitEnabled)
             partitionDataWriterContext.setWorkingDir(workingDir)
-            val metaHandler = getPartitionMetahandler(diskFileInfo)
+            val metaHandler = getPartitionMetaHandler(diskFileInfo)
             if (storageInfoType == StorageInfo.Type.HDD || storageInfoType == StorageInfo.Type.SSD) {
               new LocalTierWriter(
                 conf,