[api-draft][connector] Add simplified connector api

seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SeaTunnelRuntimeEnvironment.java

@@ -24,12 +24,7 @@
  */
 public interface SeaTunnelRuntimeEnvironment {
 
-    /**
-     * Returns the SeaTunnel runtime context.
-     *
-     * @return seaTunnelContext
-     */
-    SeaTunnelContext getSeaTunnelContext();
-
-    void setSeaTunnelContext(SeaTunnelContext seaTunnelContext);
+    default void setSeaTunnelContext(SeaTunnelContext seaTunnelContext){
+        // nothing
+    }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SeaTunnelSource.java

@@ -22,7 +22,6 @@
 import org.apache.seatunnel.api.serialization.DefaultSerializer;
 import org.apache.seatunnel.api.serialization.Serializer;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
-import org.apache.seatunnel.common.constants.JobMode;
 
 import java.io.Serializable;
 
@@ -42,10 +41,7 @@ public interface SeaTunnelSource<T, SplitT extends SourceSplit, StateT>
      *
      * @return the boundedness of this source.
      */
-    default Boundedness getBoundedness() {
-        return JobMode.BATCH.equals(getSeaTunnelContext().getJobMode()) ?
-            Boundedness.BOUNDED : Boundedness.UNBOUNDED;
-    }
+    Boundedness getBoundedness();
 
     /**
      * Get the data type of the records produced by this source.

seatunnel-connectors/seatunnel-connectors-seatunnel/pom.xml

@@ -31,6 +31,7 @@
     <artifactId>seatunnel-connectors-seatunnel</artifactId>
 
     <modules>
+        <module>seatunnel-connector-seatunnel-common</module>
         <module>seatunnel-connector-seatunnel-console</module>
         <module>seatunnel-connector-seatunnel-fake</module>
         <module>seatunnel-connector-seatunnel-kafka</module>

seatunnel-connectors/seatunnel-connectors-seatunnel/seatunnel-connector-seatunnel-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSource.java

@@ -24,7 +24,6 @@
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.Config.USERNAME;
 
 import org.apache.seatunnel.api.common.PrepareFailException;
-import org.apache.seatunnel.api.common.SeaTunnelContext;
 import org.apache.seatunnel.api.serialization.DefaultSerializer;
 import org.apache.seatunnel.api.serialization.Serializer;
 import org.apache.seatunnel.api.source.Boundedness;
@@ -58,7 +57,6 @@
 @AutoService(SeaTunnelSource.class)
 public class ClickhouseSource implements SeaTunnelSource<SeaTunnelRow, ClickhouseSourceSplit, ClickhouseSourceState> {
 
-    private SeaTunnelContext seaTunnelContext;
     private List<ClickHouseNode> servers;
     private SeaTunnelRowType rowTypeInfo;
     private String sql;
@@ -139,13 +137,4 @@ public Serializer<ClickhouseSourceState> getEnumeratorStateSerializer() {
         return new DefaultSerializer<>();
     }
 
-    @Override
-    public SeaTunnelContext getSeaTunnelContext() {
-        return seaTunnelContext;
-    }
-
-    @Override
-    public void setSeaTunnelContext(SeaTunnelContext seaTunnelContext) {
-        this.seaTunnelContext = seaTunnelContext;
-    }
 }

seatunnel-connectors/seatunnel-connectors-seatunnel/seatunnel-connector-seatunnel-common/pom.xml

@@ -0,0 +1,42 @@
+<?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-connectors-seatunnel</artifactId>
+        <groupId>org.apache.seatunnel</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>seatunnel-connector-seatunnel-common</artifactId>
+
+    <properties>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-connectors/seatunnel-connectors-seatunnel/seatunnel-connector-seatunnel-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/sink/AbstractSimpleSink.java

@@ -0,0 +1,59 @@
+/*
+ * 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.connectors.seatunnel.common.sink;
+
+import org.apache.seatunnel.api.serialization.Serializer;
+import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.SinkAggregatedCommitter;
+import org.apache.seatunnel.api.sink.SinkCommitter;
+import org.apache.seatunnel.api.sink.SinkWriter;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+
+public abstract class AbstractSimpleSink<T, StateT> implements SeaTunnelSink<T, StateT, Void, Void> {
+
+    @Override
+    public abstract AbstractSinkWriter<T, StateT> createWriter(SinkWriter.Context context) throws IOException;
+
+    @Override
+    public SinkWriter<T, Void, StateT> restoreWriter(SinkWriter.Context context, List<StateT> states) throws IOException {
+        return createWriter(context);
+    }
+
+    @Override
+    public final Optional<SinkCommitter<Void>> createCommitter() throws IOException {
+        return Optional.empty();
+    }
+
+    @Override
+    public final Optional<Serializer<Void>> getCommitInfoSerializer() {
+        return Optional.empty();
+    }
+
+    @Override
+    public final Optional<SinkAggregatedCommitter<Void, Void>> createAggregatedCommitter() throws IOException {
+        return Optional.empty();
+    }
+
+    @Override
+    public final Optional<Serializer<Void>> getAggregatedCommitInfoSerializer() {
+        return Optional.empty();
+    }
+}

seatunnel-connectors/seatunnel-connectors-seatunnel/seatunnel-connector-seatunnel-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/socket/source/SocketSourceSplit.java → seatunnel-connectors/seatunnel-connectors-seatunnel/seatunnel-connector-seatunnel-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/sink/AbstractSinkWriter.java

@@ -15,19 +15,21 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.socket.source;
+package org.apache.seatunnel.connectors.seatunnel.common.sink;
 
-import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.sink.SinkWriter;
 
-public class SocketSourceSplit implements SourceSplit {
-    private final String splitId;
+import java.util.Optional;
 
-    public SocketSourceSplit(String splitId) {
-        this.splitId = splitId;
+public abstract class AbstractSinkWriter<T, StateT> implements SinkWriter<T, Void, StateT> {
+
+    @Override
+    public final Optional<Void> prepareCommit() {
+        return Optional.empty();
     }
 
     @Override
-    public String splitId() {
-        return splitId;
+    public final void abortPrepare() {
+        // nothing
     }
 }

seatunnel-connectors/seatunnel-connectors-seatunnel/seatunnel-connector-seatunnel-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/AbstractSingleSplitReader.java

@@ -0,0 +1,67 @@
+/*
+ * 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.connectors.seatunnel.common.source;
+
+import org.apache.seatunnel.api.source.SourceEvent;
+import org.apache.seatunnel.api.source.SourceReader;
+
+import java.util.Collections;
+import java.util.List;
+
+public abstract class AbstractSingleSplitReader<T> implements SourceReader<T, SingleSplit> {
+
+    @Override
+    public final List<SingleSplit> snapshotState(long checkpointId) throws Exception {
+        return Collections.singletonList(new SingleSplit(snapshotStateToBytes(checkpointId)));
+    }
+
+    protected byte[] snapshotStateToBytes(long checkpointId) throws Exception {
+        // default nothing
+        return null;
+    }
+
+    @Override
+    public final void addSplits(List<SingleSplit> splits) {
+        if (splits.size() > 1) {
+            throw new UnsupportedOperationException("The single-split reader don't support reading multiple splits");
+        }
+        byte[] restoredState = splits.get(0).getState();
+        if (restoredState != null && restoredState.length > 0) {
+            restoreState(restoredState);
+        }
+    }
+
+    protected void restoreState(byte[] restoredState) {
+        // default nothing
+    }
+
+    @Override
+    public final void handleNoMoreSplits() {
+        // nothing
+    }
+
+    @Override
+    public void notifyCheckpointComplete(long checkpointId) throws Exception {
+        // default nothing
+    }
+
+    @Override
+    public final void handleSourceEvent(SourceEvent sourceEvent) {
+        // nothing
+    }
+}

seatunnel-connectors/seatunnel-connectors-seatunnel/seatunnel-connector-seatunnel-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/AbstractSingleSplitSource.java

@@ -0,0 +1,58 @@
+/*
+ * 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.connectors.seatunnel.common.source;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import org.apache.seatunnel.api.serialization.DefaultSerializer;
+import org.apache.seatunnel.api.serialization.Serializer;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+
+public abstract class AbstractSingleSplitSource<T> implements SeaTunnelSource<T, SingleSplit, SingleSplitEnumeratorState> {
+
+    @Override
+    public final AbstractSingleSplitReader<T> createReader(SourceReader.Context readerContext) throws Exception {
+        checkArgument(readerContext.getIndexOfSubtask() == 0, "Single split source allows only a single reader to be created.");
+        return createReader(new SingleSplitReaderContext(readerContext));
+    }
+
+    public abstract AbstractSingleSplitReader<T> createReader(SingleSplitReaderContext readerContext) throws Exception;
+
+    @Override
+    public final SourceSplitEnumerator<SingleSplit, SingleSplitEnumeratorState> createEnumerator(SourceSplitEnumerator.Context<SingleSplit> enumeratorContext) throws Exception {
+        return new SingleSplitEnumerator(enumeratorContext);
+    }
+
+    @Override
+    public final SourceSplitEnumerator<SingleSplit, SingleSplitEnumeratorState> restoreEnumerator(SourceSplitEnumerator.Context<SingleSplit> enumeratorContext, SingleSplitEnumeratorState checkpointState) throws Exception {
+        return createEnumerator(enumeratorContext);
+    }
+
+    @Override
+    public final Serializer<SingleSplitEnumeratorState> getEnumeratorStateSerializer() {
+        return new DefaultSerializer<>();
+    }
+
+    @Override
+    public final Serializer<SingleSplit> getSplitSerializer() {
+        return new DefaultSerializer<>();
+    }
+
+}

seatunnel-connectors/seatunnel-connectors-seatunnel/seatunnel-connector-seatunnel-http/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceSplit.java → seatunnel-connectors/seatunnel-connectors-seatunnel/seatunnel-connector-seatunnel-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/SingleSplit.java

@@ -15,20 +15,23 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.http.source;
+package org.apache.seatunnel.connectors.seatunnel.common.source;
 
 import org.apache.seatunnel.api.source.SourceSplit;
 
-public class HttpSourceSplit implements SourceSplit {
+public class SingleSplit implements SourceSplit {
+    private final byte[] state;
 
-    private final String splitId;
+    public SingleSplit(byte[] state) {
+        this.state = state;
+    }
 
-    public HttpSourceSplit(String splitId) {
-        this.splitId = splitId;
+    public byte[] getState() {
+        return state;
     }
 
     @Override
     public String splitId() {
-        return this.splitId;
+        return "single";
     }
 }