[Engine]Support IMap File Storage

pom.xml

@@ -568,6 +568,7 @@
                     <configuration>
                         <source>${maven.compiler.source}</source>
                         <target>${maven.compiler.target}</target>
+                        <forceJavacCompilerUse>true</forceJavacCompilerUse>
                     </configuration>
                 </plugin>
                 <plugin>

seatunnel-engine/pom.xml

@@ -28,6 +28,7 @@
     <properties>
         <!--  SeaTunnel Engine use     -->
         <hazelcast.version>5.1</hazelcast.version>
+        <disruptor.version>3.4.4</disruptor.version>
     </properties>
 
     <modules>
@@ -46,6 +47,11 @@
                 <artifactId>hazelcast</artifactId>
                 <version>${hazelcast.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.lmax</groupId>
+                <artifactId>disruptor</artifactId>
+                <version>${disruptor.version}</version>
+            </dependency>
 
             <!-- test dependencies -->
             <dependency>

seatunnel-engine/seatunnel-engine-storage/imap-storage-api/pom.xml

@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  ~
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>seatunnel-engine-storage</artifactId>
+        <groupId>org.apache.seatunnel</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>imap-storage-api</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.protostuff</groupId>
+            <artifactId>protostuff-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.protostuff</groupId>
+            <artifactId>protostuff-runtime</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/IMapStorage.java

@@ -0,0 +1,68 @@
+/*
+ * 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.api;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+public interface IMapStorage {
+
+    public void initialize(Map<String, Object> properties);
+
+    /**
+     * Store a key-value pair in the map.
+     * @param key storage key
+     * @param value storage value
+     * @return storage status, true is success, false is fail
+     */
+    public boolean store(Object key, Object value);
+
+    /**
+     * Store a key-value pair in the map storage.
+     * @param map storage key-value pair
+     * @return if some key-value pair is not stored, return this keys;
+     *         if all key-value pair is stored, return empty set.
+     */
+    public Set<Object> storeAll(Map<Object, Object> map);
+
+    /**
+     * Delete a key  in the map storage.
+     * @param key storage key
+     * @return storage status, true is success, false is fail
+     */
+    public boolean delete(Object key);
+
+    /**
+     * Delete a collection of keys from the map storage.
+     * @param  keys delete keys
+     * @return if some keys delete fail, will return this keys
+     *         if all keys delete success, will return empty set
+     */
+    public Set<Object> deleteAll(Collection<Object> keys);
+
+    public Map<Object, Object> loadAll() throws IOException;
+
+    public Set<Object> loadAllKeys();
+
+    public void destroy();
+}

seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/common/ProtoStuffSerializer.java

@@ -0,0 +1,134 @@
+/*
+ * 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.api.common;
+
+import io.protostuff.LinkedBuffer;
+import io.protostuff.ProtostuffIOUtil;
+import io.protostuff.Schema;
+import io.protostuff.runtime.RuntimeSchema;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Todo: move to common module
+ */
+@Slf4j
+public class ProtoStuffSerializer implements Serializer {
+
+    /**
+     * At the moment it looks like we only have one Schema.
+     */
+    private static final Map<Class<?>, Schema<?>> SCHEMA_CACHE = new ConcurrentHashMap<>();
+
+    @SuppressWarnings("unchecked")
+    private static <T> Schema<T> getSchema(Class<T> clazz) {
+        return (Schema<T>) SCHEMA_CACHE.computeIfAbsent(clazz, RuntimeSchema::createFrom);
+    }
+
+    private static final Set<Class<?>> WRAPPERS = new HashSet<>();
+
+    private static final Class<SerializerDeserializerWrapper> WRAPPER_CLASS = SerializerDeserializerWrapper.class;
+
+    private static final Schema<SerializerDeserializerWrapper> WRAPPER_SCHEMA = getSchema(WRAPPER_CLASS);
+
+    static {
+        WRAPPERS.add(Boolean.class);
+        WRAPPERS.add(Byte.class);
+        WRAPPERS.add(Character.class);
+        WRAPPERS.add(Short.class);
+        WRAPPERS.add(Integer.class);
+        WRAPPERS.add(Long.class);
+        WRAPPERS.add(Float.class);
+        WRAPPERS.add(Double.class);
+        WRAPPERS.add(String.class);
+        WRAPPERS.add(Void.class);
+        WRAPPERS.add(List.class);
+        WRAPPERS.add(ArrayList.class);
+        WRAPPERS.add(Map.class);
+        WRAPPERS.add(HashMap.class);
+        WRAPPERS.add(TreeMap.class);
+        WRAPPERS.add(Hashtable.class);
+        WRAPPERS.add(SortedMap.class);
+    }
+
+    @Override
+    public <T> byte[] serialize(T obj) {
+        Class<T> clazz = (Class<T>) obj.getClass();
+        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
+        Schema schema = WRAPPER_SCHEMA;
+        if (WRAPPERS.contains(clazz)) {
+            obj = (T) SerializerDeserializerWrapper.of(obj);
+        } else {
+            schema = getSchema(clazz);
+        }
+
+        byte[] data;
+        try {
+            data = ProtostuffIOUtil.toByteArray(obj, schema, buffer);
+        } finally {
+            buffer.clear();
+        }
+        return data;
+    }
+
+    @Override
+    public <T> T deserialize(byte[] data, Class<T> clz) {
+
+        if (!WRAPPERS.contains(clz)) {
+            Schema<T> schema = getSchema(clz);
+            T message = schema.newMessage();
+            ProtostuffIOUtil.mergeFrom(data, message, schema);
+            return message;
+        }
+        SerializerDeserializerWrapper<T> wrapper = new SerializerDeserializerWrapper<>();
+        ProtostuffIOUtil.mergeFrom(data, wrapper, WRAPPER_SCHEMA);
+        return wrapper.getObj();
+    }
+
+    public static class SerializerDeserializerWrapper<T> {
+        private T obj;
+
+        public static <T> SerializerDeserializerWrapper<T> of(T obj) {
+            SerializerDeserializerWrapper<T> wrapper = new SerializerDeserializerWrapper<>();
+            wrapper.setObj(obj);
+            return wrapper;
+        }
+
+        public T getObj() {
+            return obj;
+        }
+
+        public void setObj(T obj) {
+            this.obj = obj;
+        }
+    }
+
+}

seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/common/Serializer.java

@@ -0,0 +1,33 @@
+/*
+ * 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.api.common;
+
+import java.io.IOException;
+
+/**
+ * Todo: move to common module
+ */
+public interface Serializer {
+
+    <T> byte[] serialize(T obj) throws IOException;
+
+    <T> T deserialize(byte[] data, Class<T> clz) throws IOException;
+}

seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/exception/IMapStorageException.java

@@ -0,0 +1,40 @@
+/*
+ * 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.api.exception;
+
+public class IMapStorageException extends RuntimeException {
+
+    public IMapStorageException(String message) {
+        super(message);
+    }
+
+    public IMapStorageException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public IMapStorageException(Throwable cause) {
+        super(cause);
+    }
+
+    public IMapStorageException(Throwable cause, String message, Object... data) {
+        super(String.format(message, data), cause);
+    }
+}