bytedance · hk-lrzy · Nov 7, 2022 · Nov 7, 2022
diff --git a/bitsail-base/src/main/java/com/bytedance/bitsail/base/connector/reader/v1/Source.java b/bitsail-base/src/main/java/com/bytedance/bitsail/base/connector/reader/v1/Source.java
@@ -40,21 +40,39 @@ public interface Source<T, SplitT extends SourceSplit, StateT extends Serializab
    */
   Boundedness getSourceBoundedness();
 
+  /**
+   * Create Source Reader.
+   */
   SourceReader<T, SplitT> createReader(SourceReader.Context readerContext);
 
+  /**
+   * Create split coordinator.
+   */
   SourceSplitCoordinator<SplitT, StateT> createSplitCoordinator(SourceSplitCoordinator.Context<SplitT, StateT> coordinatorContext);
 
+  /**
+   * Get Split serializer for the framework,{@link SplitT}should implement from {@link  Serializable}
+   */
   default BinarySerializer<SplitT> getSplitSerializer() {
     return new SimpleBinarySerializer<>();
   }
 
-  default BinarySerializer<StateT> getEnumeratorCheckpointSerializer() {
+  /**
+   * Get State serializer for the framework, {@link StateT}should implement from {@link  Serializable}
+   */
+  default BinarySerializer<StateT> getSplitCoordinatorCheckpointSerializer() {
     return new SimpleBinarySerializer<>();
   }
 
+  /**
+   * Create type info converter for the source, default value {@link BitSailTypeInfoConverter}
+   */
   default TypeInfoConverter createTypeInfoConverter() {
     return new BitSailTypeInfoConverter();
   }
 
+  /**
+   * Get Source' name.
+   */
   String getReaderName();
 }
diff --git a/.../connector/writer/v1/WriterGenerator.java → ...itsail/base/connector/writer/v1/Sink.java b/.../connector/writer/v1/WriterGenerator.java → ...itsail/base/connector/writer/v1/Sink.java
@@ -18,20 +18,20 @@
 package com.bytedance.bitsail.base.connector.writer.v1;
 
 import com.bytedance.bitsail.base.serializer.BinarySerializer;
+import com.bytedance.bitsail.base.serializer.SimpleBinarySerializer;
 import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
 import com.bytedance.bitsail.common.type.BitSailTypeInfoConverter;
 import com.bytedance.bitsail.common.type.TypeInfoConverter;
 import com.bytedance.bitsail.common.typeinfo.TypeInfo;
 
 import java.io.IOException;
 import java.io.Serializable;
-import java.util.List;
 import java.util.Optional;
 
 /**
  * Created 2022/6/10
  */
-public interface WriterGenerator<InputT, CommitT, WriterStateT> extends Serializable {
+public interface Sink<InputT, CommitT extends Serializable, WriterStateT extends Serializable> extends Serializable {
 
   /**
    * @return The name of writer operation.
@@ -49,23 +49,9 @@ public interface WriterGenerator<InputT, CommitT, WriterStateT> extends Serializ
   /**
    * Create a writer for processing elements.
    *
-   * @param writerConfiguration Options for writer.
    * @return An initialized writer.
    */
-  Writer<InputT, CommitT, WriterStateT> createWriter(BitSailConfiguration writerConfiguration,
-                                                     Writer.Context context) throws IOException;
-
-  /**
-   * Restore a writer for processing elements from configurations and history states.
-   *
-   * @param writerConfiguration Options for writer.
-   * @param writerStates        History writer states.
-   * @param context             Writer context
-   * @return An initialized writer.
-   */
-  Writer<InputT, CommitT, WriterStateT> restoreWriter(BitSailConfiguration writerConfiguration,
-                                                      List<WriterStateT> writerStates,
-                                                      Writer.Context context) throws IOException;
+  Writer<InputT, CommitT, WriterStateT> createWriter(Writer.Context<WriterStateT> context) throws IOException;
 
   /**
    * @return A converter which supports conversion from BitSail {@link TypeInfo}
@@ -85,14 +71,14 @@ default Optional<WriterCommitter<CommitT>> createCommitter() {
   /**
    * @return A serializer which convert committable object to byte array.
    */
-  default Optional<BinarySerializer<CommitT>> getCommittableSerializer() {
-    return Optional.empty();
+  default BinarySerializer<CommitT> getCommittableSerializer() {
+    return new SimpleBinarySerializer<CommitT>();
   }
 
   /**
    * @return A serializer which convert state object to byte array.
    */
-  default Optional<BinarySerializer<WriterStateT>> getWriteStateSerializer() {
-    return Optional.empty();
+  default BinarySerializer<WriterStateT> getWriteStateSerializer() {
+    return new SimpleBinarySerializer<WriterStateT>();
   }
 }
diff --git a/bitsail-base/src/main/java/com/bytedance/bitsail/base/connector/writer/v1/Writer.java b/bitsail-base/src/main/java/com/bytedance/bitsail/base/connector/writer/v1/Writer.java
@@ -72,10 +72,14 @@ default void close() throws IOException {
 
   }
 
-  interface Context extends Serializable {
+  interface Context<WriterStateT> extends Serializable {
 
     TypeInfo<?>[] getTypeInfos();
 
     int getIndexOfSubTaskId();
+
+    boolean isRestored();
+
+    List<WriterStateT> getRestoreStates();
   }
 }
diff --git a/...main/java/com/bytedance/bitsail/base/connector/writer/v1/context/SimpleWriterContext.java b/...main/java/com/bytedance/bitsail/base/connector/writer/v1/context/SimpleWriterContext.java
diff --git a/...-fake/src/main/resources/legacy_fake.json → ...ources/bitsail-connector-legacy-fake.json b/...-fake/src/main/resources/legacy_fake.json → ...ources/bitsail-connector-legacy-fake.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-fake",
+  "name": "bitsail-connector-legacy-fake",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.fake.source.FakeSource",
     "com.bytedance.bitsail.connector.legacy.fake.source.FakeFlinkSource"

diff --git a/...or-ftp/src/main/resources/legacy_ftp.json → ...sources/bitsail-connector-legacy-ftp.json b/...or-ftp/src/main/resources/legacy_ftp.json → ...sources/bitsail-connector-legacy-ftp.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-ftp",
+  "name": "bitsail-connector-legacy-ftp",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.ftp.source.FtpInputFormat"
   ],

diff --git a/...oop/src/main/resources/legacy_hadoop.json → ...rces/bitsail-connector-legacy-hadoop.json b/...oop/src/main/resources/legacy_hadoop.json → ...rces/bitsail-connector-legacy-hadoop.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-hadoop",
+  "name": "bitsail-connector-legacy-hadoop",
   "classes": [
     "com.bytedance.bitsail.connector.hadoop.source.HadoopInputFormat"
   ],

diff --git a/...base/src/main/resources/legacy_hbase.json → ...urces/bitsail-connector-legacy-hbase.json b/...base/src/main/resources/legacy_hbase.json → ...urces/bitsail-connector-legacy-hbase.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-hbase",
+  "name": "bitsail-connector-legacy-hbase",
   "classes": [
     "com.bytedance.bitsail.connector.hbase.source.HBaseInputFormat",
     "com.bytedance.bitsail.connector.hbase.sink.HBaseOutputFormat"

diff --git a/...-hive/src/main/resources/legacy_hive.json → ...ources/bitsail-connector-legacy-hive.json b/...-hive/src/main/resources/legacy_hive.json → ...ources/bitsail-connector-legacy-hive.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-hive",
+  "name": "bitsail-connector-legacy-hive",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.hive.source.HiveInputFormat",
     "com.bytedance.bitsail.connector.legacy.hive.sink.HiveOutputFormat"

diff --git a/...-hudi/src/main/resources/legacy_hudi.json → ...ources/bitsail-connector-legacy-hudi.json b/...-hudi/src/main/resources/legacy_hudi.json → ...ources/bitsail-connector-legacy-hudi.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-hudi",
+  "name": "bitsail-connector-legacy-hudi",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.hudi.dag.HudiSinkFunctionDAGBuilder",
     "com.bytedance.bitsail.connector.legacy.hudi.dag.HudiSourceFunctionDAGBuilder",

diff --git a/...-jdbc/src/main/resources/legacy_jdbc.json → ...ources/bitsail-connector-legacy-jdbc.json b/...-jdbc/src/main/resources/legacy_jdbc.json → ...ources/bitsail-connector-legacy-jdbc.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-jdbc",
+  "name": "bitsail-connector-legacy-jdbc",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.jdbc.sink.JDBCOutputFormat",
     "com.bytedance.bitsail.connector.legacy.jdbc.source.JDBCInputFormat",

diff --git a/...afka/src/main/resources/legacy_kafka.json → ...urces/bitsail-connector-legacy-kafka.json b/...afka/src/main/resources/legacy_kafka.json → ...urces/bitsail-connector-legacy-kafka.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-kafka",
+  "name": "bitsail-connector-legacy-kafka",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.kafka.source.KafkaSourceFunctionDAGBuilder",
     "com.bytedance.bitsail.connector.legacy.kafka.sink.KafkaOutputFormat"

diff --git a/.../src/main/resources/legacy_larksheet.json → ...s/bitsail-connector-legacy-larksheet.json b/.../src/main/resources/legacy_larksheet.json → ...s/bitsail-connector-legacy-larksheet.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-larksheet",
+  "name": "bitsail-connector-legacy-larksheet",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.larksheet.source.LarkSheetInputFormat"
   ],

diff --git a/...db/src/main/resources/legacy_mongodb.json → ...ces/bitsail-connector-legacy-mongodb.json b/...db/src/main/resources/legacy_mongodb.json → ...ces/bitsail-connector-legacy-mongodb.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-mongodb",
+  "name": "bitsail-connector-legacy-mongodb",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.mongodb.source.MongoDBInputFormat",
     "com.bytedance.bitsail.connector.legacy.mongodb.sink.MongoDBOutputFormat"

diff --git a/...rint/src/main/resources/legacy_print.json → ...urces/bitsail-connector-legacy-print.json b/...rint/src/main/resources/legacy_print.json → ...urces/bitsail-connector-legacy-print.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-print",
+  "name": "bitsail-connector-legacy-print",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.print.sink.PrintSink"
   ],

diff --git a/...edis/src/main/resources/legacy_redis.json → ...urces/bitsail-connector-legacy-redis.json b/...edis/src/main/resources/legacy_redis.json → ...urces/bitsail-connector-legacy-redis.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-redis",
+  "name": "bitsail-connector-legacy-redis",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.redis.source.RedisOutputFormat"
   ],

diff --git a/.../main/resources/legacy_streamingfile.json → ...tsail-connector-legacy-streamingfile.json b/.../main/resources/legacy_streamingfile.json → ...tsail-connector-legacy-streamingfile.json
@@ -1,5 +1,5 @@
 {
-  "name": "bitsail-connector-streamingfile",
+  "name": "bitsail-connector-legacy-streamingfile",
   "classes": [
     "com.bytedance.bitsail.connector.legacy.streamingfile.sink.FileSystemSinkFunctionDAGBuilder"
   ],

diff --git a/...ctor/doris/sink/DorisWriterGenerator.java → ...tsail/connector/doris/sink/DorisSink.java b/...ctor/doris/sink/DorisWriterGenerator.java → ...tsail/connector/doris/sink/DorisSink.java
@@ -17,9 +17,9 @@
 
 package com.bytedance.bitsail.connector.doris.sink;
 
+import com.bytedance.bitsail.base.connector.writer.v1.Sink;
 import com.bytedance.bitsail.base.connector.writer.v1.Writer;
 import com.bytedance.bitsail.base.connector.writer.v1.WriterCommitter;
-import com.bytedance.bitsail.base.connector.writer.v1.WriterGenerator;
 import com.bytedance.bitsail.base.serializer.BinarySerializer;
 import com.bytedance.bitsail.base.serializer.SimpleBinarySerializer;
 import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
@@ -52,11 +52,12 @@
 import java.util.Properties;
 import java.util.stream.Collectors;
 
-public class DorisWriterGenerator<InputT> implements WriterGenerator<InputT, DorisCommittable, DorisWriterState> {
-  private static final Logger LOG = LoggerFactory.getLogger(DorisWriterGenerator.class);
+public class DorisSink<InputT> implements Sink<InputT, DorisCommittable, DorisWriterState> {
+  private static final Logger LOG = LoggerFactory.getLogger(DorisSink.class);
   private DorisExecutionOptions.WRITE_MODE writeMode;
   private DorisOptions dorisOptions;
   private DorisExecutionOptions dorisExecutionOptions;
+  private BitSailConfiguration writerConfiguration;
 
   @Override
   public String getWriterName() {
@@ -65,6 +66,7 @@ public String getWriterName() {
 
   @Override
   public void configure(BitSailConfiguration commonConfiguration, BitSailConfiguration writerConfiguration) throws IOException {
+    this.writerConfiguration = writerConfiguration;
     this.writeMode =
         DorisExecutionOptions.WRITE_MODE.valueOf(writerConfiguration.get(DorisWriterOptions.SINK_WRITE_MODE).toUpperCase());
     initDorisExecutionOptions(writerConfiguration);
@@ -86,15 +88,10 @@ public void configure(BitSailConfiguration commonConfiguration, BitSailConfigura
   }
 
   @Override
-  public Writer<InputT, DorisCommittable, DorisWriterState> createWriter(BitSailConfiguration writerConfiguration, Writer.Context context) {
+  public Writer<InputT, DorisCommittable, DorisWriterState> createWriter(Writer.Context<DorisWriterState> context) {
     return new DorisWriter(writerConfiguration, dorisOptions, dorisExecutionOptions);
   }
 
-  @Override
-  public Writer<InputT, DorisCommittable, DorisWriterState> restoreWriter(BitSailConfiguration writerConfiguration, List writerStates, Writer.Context context) {
-    return createWriter(writerConfiguration, null);
-  }
-
   @Override
   public TypeInfoConverter createTypeInfoConverter() {
     return new FileMappingTypeInfoConverter(getWriterName());
@@ -106,13 +103,13 @@ public Optional<WriterCommitter<DorisCommittable>> createCommitter() {
   }
 
   @Override
-  public Optional<BinarySerializer<DorisCommittable>> getCommittableSerializer() {
-    return Optional.of(new DorisCommittableSerializer());
+  public BinarySerializer<DorisCommittable> getCommittableSerializer() {
+    return new DorisCommittableSerializer();
   }
 
   @Override
-  public Optional<BinarySerializer<DorisWriterState>> getWriteStateSerializer() {
-    return Optional.of(new SimpleBinarySerializer<DorisWriterState>());
+  public BinarySerializer<DorisWriterState> getWriteStateSerializer() {
+    return new SimpleBinarySerializer<DorisWriterState>();
   }
 
   private void initDorisOptions(BitSailConfiguration writerConfiguration) {

diff --git a/bitsail-connectors/connector-doris/src/main/resources/bitsail-connector-unified-doris.json b/bitsail-connectors/connector-doris/src/main/resources/bitsail-connector-unified-doris.json
@@ -0,0 +1,9 @@
+{
+  "name": "bitsail-connector-unified-doris",
+  "classes": [
+    "com.bytedance.bitsail.connector.doris.sink.DorisSink"
+  ],
+  "libs": [
+    "bitsail-connector-doris-${version}.jar"
+  ]
+}
diff --git a/bitsail-connectors/connector-doris/src/main/resources/unified_doris.json b/bitsail-connectors/connector-doris/src/main/resources/unified_doris.json