code style

TongchengOpenSource · Mar 31, 2024 · ba93f88 · ba93f88
1 parent 9473c97
commit ba93f88
Show file tree

Hide file tree

Showing 7 changed files with 31 additions and 44 deletions.
diff --git a/...age-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorage.java b/...age-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorage.java
@@ -25,7 +25,7 @@
 import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
 import org.apache.seatunnel.engine.imap.storage.file.common.FileConstants;
 import org.apache.seatunnel.engine.imap.storage.file.common.WALReader;
-import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncMethod;
+import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncType;
 import org.apache.seatunnel.engine.imap.storage.file.config.AbstractConfiguration;
 import org.apache.seatunnel.engine.imap.storage.file.config.FileConfiguration;
 import org.apache.seatunnel.engine.imap.storage.file.disruptor.WALDisruptor;
@@ -72,7 +72,7 @@ public class IMapFileStorage implements IMapStorage {
 
     private static final String STORAGE_TYPE_KEY = "storage.type";
 
-    private static final String WAL_SYNC_METHOD_KEY = "wal.sync.method";
+    private static final String WAL_SYNC_TYPE_KEY = "wal.sync.type";
 
     public FileSystem fs;
 
@@ -110,7 +110,7 @@ public class IMapFileStorage implements IMapStorage {
     public static final long DEFAULT_WRITE_DATA_TIMEOUT_MILLISECONDS = 1000 * 60;
 
     private Configuration conf;
-    private WALSyncMethod walSyncMethod;
+    private WALSyncType walSyncType;
 
     private FileConfiguration fileConfiguration;
 
@@ -125,13 +125,13 @@ public void initialize(Map<String, Object> configuration) {
         String storageType =
                 String.valueOf(
                         configuration.getOrDefault(
-                                STORAGE_TYPE_KEY, FileConfiguration.HDFS.toString()));
+                                STORAGE_TYPE_KEY, FileConfiguration.HDFS.toString())).toUpperCase();
         String walSyncMethod =
                 String.valueOf(
                         configuration.getOrDefault(
-                                WAL_SYNC_METHOD_KEY, WALSyncMethod.SYNC.toString()));
-        this.walSyncMethod = WALSyncMethod.valueOf(walSyncMethod.toUpperCase());
-        this.fileConfiguration = FileConfiguration.valueOf(storageType.toUpperCase());
+                                WAL_SYNC_TYPE_KEY, WALSyncType.SYNC.toString())).toUpperCase();
+        this.walSyncType = WALSyncType.valueOf(walSyncMethod);
+        this.fileConfiguration = FileConfiguration.valueOf(storageType);
         // build configuration
         AbstractConfiguration fileConfiguration = this.fileConfiguration.getConfiguration();
 
@@ -165,8 +165,8 @@ public void initialize(Map<String, Object> configuration) {
         this.walDisruptor =
                 new WALDisruptor(
                         fs,
-                        FileConfiguration.valueOf(storageType.toUpperCase()),
-                        WALSyncMethod.valueOf(walSyncMethod.toUpperCase()),
+                        FileConfiguration.valueOf(storageType),
+                        WALSyncType.valueOf(walSyncMethod),
                         businessRootPath + region + DEFAULT_IMAP_FILE_PATH_SPLIT,
                         serializer);
     }
@@ -200,9 +200,6 @@ public Set<Object> storeAll(Map<Object, Object> map) {
                         failures.add(key);
                     }
                 });
-        if (WALSyncMethod.ASYNC == walSyncMethod) {
-            return failures;
-        }
         return batchQueryExecuteFailsStatus(requestMap, failures);
     }
 
@@ -319,7 +316,7 @@ private long sendToDisruptorQueue(IMapFileData data, WALEventType type) {
     }
 
     private boolean queryExecuteStatus(long requestId) {
-        if (WALSyncMethod.ASYNC == walSyncMethod) {
+        if (WALSyncType.ASYNC == walSyncType) {
             return true;
         }
         return queryExecuteStatus(requestId, this.writDataTimeoutMilliseconds);
@@ -343,17 +340,8 @@ private boolean queryExecuteStatus(long requestId, long timeout) {
     private Set<Object> batchQueryExecuteFailsStatus(
             Map<Long, Object> requestMap, Set<Object> failures) {
         for (Map.Entry<Long, Object> entry : requestMap.entrySet()) {
-            boolean success = false;
-            RequestFuture requestFuture = RequestFutureCache.get(entry.getKey());
-            try {
-                if (requestFuture.isDone() || Boolean.TRUE.equals(requestFuture.get())) {
-                    success = true;
-                }
-            } catch (Exception e) {
-                log.error("wait for write status error", e);
-            } finally {
-                RequestFutureCache.remove(entry.getKey());
-            }
+            Long requestId = entry.getKey();
+            boolean success = queryExecuteStatus(requestId);
             if (!success) {
                 failures.add(entry.getValue());
             }

diff --git a/...ap/storage/file/common/WALSyncMethod.java → ...imap/storage/file/common/WALSyncType.java b/...ap/storage/file/common/WALSyncMethod.java → ...imap/storage/file/common/WALSyncType.java
@@ -20,7 +20,7 @@
 
 package org.apache.seatunnel.engine.imap.storage.file.common;
 
-public enum WALSyncMethod {
+public enum WALSyncType {
     SYNC,
     ASYNC
 }
diff --git a/...ge-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALWriter.java b/...ge-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALWriter.java
@@ -56,20 +56,20 @@ public class WALWriter implements AutoCloseable {
 
     private static final int DEFAULT_THREAD_POOL_QUENE_SIZE = 1024;
 
-    private WALSyncMethod walSyncMethod;
+    private final WALSyncType walSyncType;
 
     public WALWriter(
             FileSystem fs,
             FileConfiguration fileConfiguration,
-            WALSyncMethod walSyncMethod,
+            WALSyncType walSyncType,
             Path parentPath,
             Serializer serializer)
             throws IOException {
         this.writer = DiscoveryWalFileFactory.getWriter(fileConfiguration.getName());
         this.writer.setBlockSize(fileConfiguration.getConfiguration().getBlockSize());
         this.writer.initialize(fs, parentPath, serializer);
-        this.walSyncMethod = walSyncMethod;
-        if (WALSyncMethod.ASYNC == walSyncMethod) {
+        this.walSyncType = walSyncType;
+        if (WALSyncType.ASYNC == walSyncType) {
             this.walWriterService =
                     new ThreadPoolExecutor(
                             DEFAULT_THREAD_POOL_MIN_SIZE,
@@ -83,7 +83,7 @@ public WALWriter(
 
     public void write(IMapFileData data) throws IOException {
 
-        switch (walSyncMethod) {
+        switch (walSyncType) {
             case SYNC:
                 this.writer.write(data);
                 return;
@@ -103,11 +103,10 @@ public void write(IMapFileData data) throws IOException {
 
     @Override
     public void close() throws Exception {
-        if (WALSyncMethod.ASYNC == walSyncMethod && writeTaskFuture != null) {
-            writeTaskFuture.get();
+        if (WALSyncType.ASYNC == walSyncType && writeTaskFuture != null) {
             writeTaskFuture.cancel(false);
             if (walWriterService != null) {
-                walWriterService.shutdownNow();
+                walWriterService.shutdown();
             }
         }
         this.writer.close();

diff --git a/...e/src/main/java/org/apache/seatunnel/engine/imap/storage/file/disruptor/WALDisruptor.java b/...e/src/main/java/org/apache/seatunnel/engine/imap/storage/file/disruptor/WALDisruptor.java
@@ -22,7 +22,7 @@
 
 import org.apache.seatunnel.engine.imap.storage.api.exception.IMapStorageException;
 import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
-import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncMethod;
+import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncType;
 import org.apache.seatunnel.engine.imap.storage.file.config.FileConfiguration;
 import org.apache.seatunnel.engine.serializer.api.Serializer;
 
@@ -63,7 +63,7 @@ public class WALDisruptor implements Closeable {
     public WALDisruptor(
             FileSystem fs,
             FileConfiguration fileConfiguration,
-            WALSyncMethod walSyncMethod,
+            WALSyncType walSyncType,
             String parentPath,
             Serializer serializer) {
         // todo should support multi thread producer
@@ -76,7 +76,7 @@ public WALDisruptor(
                         ProducerType.SINGLE,
                         new BlockingWaitStrategy());
         disruptor.handleEventsWithWorkerPool(
-                new WALWorkHandler(fs, fileConfiguration, walSyncMethod, parentPath, serializer));
+                new WALWorkHandler(fs, fileConfiguration, walSyncType, parentPath, serializer));
 
         disruptor.start();
     }

diff --git a/...src/main/java/org/apache/seatunnel/engine/imap/storage/file/disruptor/WALWorkHandler.java b/...src/main/java/org/apache/seatunnel/engine/imap/storage/file/disruptor/WALWorkHandler.java
@@ -22,7 +22,7 @@
 
 import org.apache.seatunnel.engine.imap.storage.api.exception.IMapStorageException;
 import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
-import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncMethod;
+import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncType;
 import org.apache.seatunnel.engine.imap.storage.file.common.WALWriter;
 import org.apache.seatunnel.engine.imap.storage.file.config.FileConfiguration;
 import org.apache.seatunnel.engine.imap.storage.file.future.RequestFutureCache;
@@ -45,13 +45,13 @@ public class WALWorkHandler implements WorkHandler<FileWALEvent> {
     public WALWorkHandler(
             FileSystem fs,
             FileConfiguration fileConfiguration,
-            WALSyncMethod walSyncMethod,
+            WALSyncType walSyncType,
             String parentPath,
             Serializer serializer) {
         try {
             writer =
                     new WALWriter(
-                            fs, fileConfiguration, walSyncMethod, new Path(parentPath), serializer);
+                            fs, fileConfiguration, walSyncType, new Path(parentPath), serializer);
         } catch (IOException e) {
             throw new IMapStorageException(
                     e, "create new current writer failed, parent path is %s", parentPath);

diff --git a/...est/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java b/...est/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
@@ -62,7 +62,7 @@ public static void init() throws IOException {
     public void testSyncWriterAndReader() throws Exception {
         WALWriter writer =
                 new WALWriter(
-                        FS, FileConfiguration.HDFS, WALSyncMethod.SYNC, PARENT_PATH, SERIALIZER);
+                        FS, FileConfiguration.HDFS, WALSyncType.SYNC, PARENT_PATH, SERIALIZER);
         IMapFileData data;
         boolean isDelete;
         for (int i = 0; i < 1024; i++) {
@@ -120,7 +120,7 @@ public void testSyncWriterAndReader() throws Exception {
     public void testAsyncWriterAndReader() throws Exception {
         WALWriter writer =
                 new WALWriter(
-                        FS, FileConfiguration.HDFS, WALSyncMethod.ASYNC, PARENT_PATH, SERIALIZER);
+                        FS, FileConfiguration.HDFS, WALSyncType.ASYNC, PARENT_PATH, SERIALIZER);
         IMapFileData data;
         boolean isDelete;
         for (int i = 0; i < 1024; i++) {

diff --git a/...c/test/java/org/apache/seatunnel/engine/imap/storage/file/disruptor/WALDisruptorTest.java b/...c/test/java/org/apache/seatunnel/engine/imap/storage/file/disruptor/WALDisruptorTest.java
@@ -21,7 +21,7 @@
 package org.apache.seatunnel.engine.imap.storage.file.disruptor;
 
 import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
-import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncMethod;
+import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncType;
 import org.apache.seatunnel.engine.imap.storage.file.config.FileConfiguration;
 import org.apache.seatunnel.engine.imap.storage.file.future.RequestFuture;
 import org.apache.seatunnel.engine.imap.storage.file.future.RequestFutureCache;
@@ -65,7 +65,7 @@ void testSyncProducerAndConsumer() throws IOException {
                 new WALDisruptor(
                         FS,
                         FileConfiguration.HDFS,
-                        WALSyncMethod.SYNC,
+                        WALSyncType.SYNC,
                         FILEPATH,
                         new ProtoStuffSerializer());
         IMapFileData data;
@@ -93,7 +93,7 @@ void testAsyncProducerAndConsumer() throws IOException {
                 new WALDisruptor(
                         FS,
                         FileConfiguration.HDFS,
-                        WALSyncMethod.ASYNC,
+                        WALSyncType.ASYNC,
                         FILEPATH,
                         new ProtoStuffSerializer());
         IMapFileData data;