add license header

TongchengOpenSource · Mar 30, 2024 · 98fb34b · 98fb34b
1 parent 206b49f
commit 98fb34b
Show file tree

Hide file tree

Showing 8 changed files with 52 additions and 31 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.WALWriteType;
+import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncMethod;
 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_TYPE_KEY = "wal.type";
+    private static final String WAL_SYNC_METHOD_KEY = "wal.sync.method";
 
     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 WALWriteType walWriteType;
+    private WALSyncMethod walSyncMethod;
 
     private FileConfiguration fileConfiguration;
 
@@ -126,10 +126,11 @@ public void initialize(Map<String, Object> configuration) {
                 String.valueOf(
                         configuration.getOrDefault(
                                 STORAGE_TYPE_KEY, FileConfiguration.HDFS.toString()));
-        String walType =
+        String walSyncMethod =
                 String.valueOf(
-                        configuration.getOrDefault(WAL_TYPE_KEY, WALWriteType.SYNC.toString()));
-        this.walWriteType = WALWriteType.valueOf(walType.toUpperCase());
+                        configuration.getOrDefault(
+                                WAL_SYNC_METHOD_KEY, WALSyncMethod.SYNC.toString()));
+        this.walSyncMethod = WALSyncMethod.valueOf(walSyncMethod.toUpperCase());
         this.fileConfiguration = FileConfiguration.valueOf(storageType.toUpperCase());
         // build configuration
         AbstractConfiguration fileConfiguration = this.fileConfiguration.getConfiguration();
@@ -165,7 +166,7 @@ public void initialize(Map<String, Object> configuration) {
                 new WALDisruptor(
                         fs,
                         FileConfiguration.valueOf(storageType.toUpperCase()),
-                        WALWriteType.valueOf(walType.toUpperCase()),
+                        WALSyncMethod.valueOf(walSyncMethod.toUpperCase()),
                         businessRootPath + region + DEFAULT_IMAP_FILE_PATH_SPLIT,
                         serializer);
     }
@@ -315,7 +316,7 @@ private long sendToDisruptorQueue(IMapFileData data, WALEventType type) {
     }
 
     private boolean queryExecuteStatus(long requestId) {
-        if (WALWriteType.ASYNC == walWriteType) {
+        if (WALSyncMethod.ASYNC == walSyncMethod) {
             return true;
         }
         return queryExecuteStatus(requestId, this.writDataTimeoutMilliseconds);

diff --git a/...ile/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALSyncMethod.java b/...ile/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALSyncMethod.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.seatunnel.engine.imap.storage.file.common;
+
+public enum WALSyncMethod {
+    SYNC,
+    ASYNC
+}
diff --git a/...file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALWriteType.java b/...file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALWriteType.java
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 WALWriteType walWriteType;
+    private WALSyncMethod walSyncMethod;
 
     public WALWriter(
             FileSystem fs,
             FileConfiguration fileConfiguration,
-            WALWriteType walWriteType,
+            WALSyncMethod walSyncMethod,
             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.walWriteType = walWriteType;
-        if (WALWriteType.ASYNC == walWriteType) {
+        this.walSyncMethod = walSyncMethod;
+        if (WALSyncMethod.ASYNC == walSyncMethod) {
             this.walWriterService =
                     new ThreadPoolExecutor(
                             DEFAULT_THREAD_POOL_MIN_SIZE,
@@ -83,7 +83,7 @@ public WALWriter(
 
     public void write(IMapFileData data) throws IOException {
 
-        switch (walWriteType) {
+        switch (walSyncMethod) {
             case SYNC:
                 this.writer.write(data);
                 return;
@@ -103,7 +103,7 @@ public void write(IMapFileData data) throws IOException {
 
     @Override
     public void close() throws Exception {
-        if (WALWriteType.ASYNC == walWriteType && writeTaskFuture != null) {
+        if (WALSyncMethod.ASYNC == walSyncMethod && writeTaskFuture != null) {
             writeTaskFuture.get();
             writeTaskFuture.cancel(false);
             if (walWriterService != null) {

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.WALWriteType;
+import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncMethod;
 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,
-            WALWriteType walWriteType,
+            WALSyncMethod walSyncMethod,
             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, walWriteType, parentPath, serializer));
+                new WALWorkHandler(fs, fileConfiguration, walSyncMethod, 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.WALWriteType;
+import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncMethod;
 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,
-            WALWriteType walWriteType,
+            WALSyncMethod walSyncMethod,
             String parentPath,
             Serializer serializer) {
         try {
             writer =
                     new WALWriter(
-                            fs, fileConfiguration, walWriteType, new Path(parentPath), serializer);
+                            fs, fileConfiguration, walSyncMethod, 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, WALWriteType.SYNC, PARENT_PATH, SERIALIZER);
+                        FS, FileConfiguration.HDFS, WALSyncMethod.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, WALWriteType.ASYNC, PARENT_PATH, SERIALIZER);
+                        FS, FileConfiguration.HDFS, WALSyncMethod.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.WALWriteType;
+import org.apache.seatunnel.engine.imap.storage.file.common.WALSyncMethod;
 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,
-                        WALWriteType.SYNC,
+                        WALSyncMethod.SYNC,
                         FILEPATH,
                         new ProtoStuffSerializer());
         IMapFileData data;
@@ -93,7 +93,7 @@ void testAsyncProducerAndConsumer() throws IOException {
                 new WALDisruptor(
                         FS,
                         FileConfiguration.HDFS,
-                        WALWriteType.ASYNC,
+                        WALSyncMethod.ASYNC,
                         FILEPATH,
                         new ProtoStuffSerializer());
         IMapFileData data;