[BitSail][CDC] Add cdc source coordinator unit test (#407)

Author: garyli1019

bitsail-connectors/connector-cdc/connector-cdc-base/src/main/java/com/bytedance/bitsail/connector/cdc/source/coordinator/CDCSourceSplitCoordinator.java

@@ -58,8 +58,6 @@ public CDCSourceSplitCoordinator(SourceSplitCoordinator.Context<BinlogSplit, Bas
 
   @Override
   public void start() {
-    int totalReader = this.context.registeredReaders().size();
-    LOG.info("Total registered reader number: {}", totalReader);
     if (!this.isBinlogAssigned) {
       List<BinlogSplit> splitList = new ArrayList<>();
       BinlogSplit split = createSplit(this.jobConf);

bitsail-connectors/connector-cdc/connector-cdc-base/src/main/java/com/bytedance/bitsail/connector/cdc/source/offset/BinlogOffset.java

@@ -57,8 +57,7 @@ public static BinlogOffset specified() {
   }
 
   public static BinlogOffset createFromJobConf(BitSailConfiguration jobConf) {
-    String rawOffsetType = jobConf.getNecessaryOption(
-        BinlogReaderOptions.INITIAL_OFFSET_TYPE, BinlogReaderErrorCode.REQUIRED_VALUE).toUpperCase().trim();
+    String rawOffsetType = jobConf.get(BinlogReaderOptions.INITIAL_OFFSET_TYPE).toUpperCase().trim();
     BinlogOffsetType offsetType = BinlogOffsetType.valueOf(rawOffsetType);
     switch (offsetType) {
       case LATEST:

bitsail-connectors/connector-cdc/connector-cdc-base/src/test/java/com/bytedance/bitsail/connector/cdc/source/coordinator/CDCSourceSplitCoordinatorTests.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2022-2023 Bytedance Ltd. and/or its affiliates.
+ *
+ * Licensed 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 com.bytedance.bitsail.connector.cdc.source.coordinator;
+
+import com.bytedance.bitsail.base.connector.reader.v1.SourceEvent;
+import com.bytedance.bitsail.base.connector.reader.v1.SourceSplitCoordinator;
+import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
+import com.bytedance.bitsail.connector.cdc.source.coordinator.state.BaseAssignmentState;
+import com.bytedance.bitsail.connector.cdc.source.coordinator.state.BinlogAssignmentState;
+import com.bytedance.bitsail.connector.cdc.source.offset.BinlogOffsetType;
+import com.bytedance.bitsail.connector.cdc.source.split.BinlogSplit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.function.BiConsumer;
+
+public class CDCSourceSplitCoordinatorTests {
+  @Test
+  public void testStartNew() {
+    BitSailConfiguration jobConf = BitSailConfiguration.newDefault();
+    Map<Integer, List<BinlogSplit>> assigned = new HashMap<Integer, List<BinlogSplit>>();
+    SourceSplitCoordinator.Context context = getContext(false, null, assigned);
+    CDCSourceSplitCoordinator coordinator = new CDCSourceSplitCoordinator(context, jobConf);
+    coordinator.start();
+    Assert.assertEquals(1, assigned.get(0).size());
+    Assert.assertEquals(BinlogOffsetType.EARLIEST, assigned.get(0).get(0).getBeginOffset().getOffsetType());
+    BinlogAssignmentState state = (BinlogAssignmentState) coordinator.snapshotState();
+    Assert.assertTrue(state.isAssigned());
+  }
+
+  @Test
+  public void testRestore() {
+    BitSailConfiguration jobConf = BitSailConfiguration.newDefault();
+    Map<Integer, List<BinlogSplit>> assigned = new HashMap<Integer, List<BinlogSplit>>();
+    BinlogAssignmentState state = new BinlogAssignmentState(true);
+    SourceSplitCoordinator.Context context = getContext(true, state, assigned);
+    CDCSourceSplitCoordinator coordinator = new CDCSourceSplitCoordinator(context, jobConf);
+    coordinator.start();
+    // no split was assigned
+    Assert.assertEquals(0, assigned.size());
+    BinlogAssignmentState newState = (BinlogAssignmentState) coordinator.snapshotState();
+    Assert.assertTrue(newState.isAssigned());
+  }
+
+  SourceSplitCoordinator.Context<BinlogSplit, BaseAssignmentState> getContext(
+      boolean isRestore, BaseAssignmentState restoredState, Map<Integer, List<BinlogSplit>> assigned) {
+    SourceSplitCoordinator.Context<BinlogSplit, BaseAssignmentState> context = new SourceSplitCoordinator.Context<BinlogSplit, BaseAssignmentState>() {
+      @Override
+      public boolean isRestored() {
+        return isRestore;
+      }
+
+      @Override
+      public BaseAssignmentState getRestoreState() {
+        return restoredState;
+      }
+
+      @Override
+      public int totalParallelism() {
+        return 1;
+      }
+
+      @Override
+      public Set<Integer> registeredReaders() {
+        return null;
+      }
+
+      @Override
+      public void assignSplit(int subtaskId, List splits) {
+        assigned.put(subtaskId, splits);
+      }
+
+      @Override
+      public void signalNoMoreSplits(int subtask) {
+
+      }
+
+      @Override
+      public void sendEventToSourceReader(int subtaskId, SourceEvent event) {
+
+      }
+
+      @Override
+      public void runAsync(Callable callable, BiConsumer handler, int initialDelay, long interval) {
+
+      }
+
+      @Override
+      public void runAsyncOnce(Callable callable, BiConsumer handler) {
+
+      }
+    };
+    return context;
+  }
+}

bitsail-connectors/connector-cdc/connector-cdc-mysql/src/test/java/com/bytedance/bitsail/connector/cdc/mysql/source/reader/MysqlBinlogSourceReaderTest.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2022-2023 Bytedance Ltd. and/or its affiliates.
+ *
+ * Licensed 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 com.bytedance.bitsail.connector.cdc.mysql.source.reader;
+
+import org.junit.Test;
+
+public class MysqlBinlogSourceReaderTest {
+  @Test
+  public void testReader() {
+
+  }
+}