From 0bdc72de2a5df09e3bfe53d472b0f479e7a83ba4 Mon Sep 17 00:00:00 2001
From: tyrantlucifer <tyrantlucifer@gmail.com>
Date: Tue, 21 Feb 2023 19:20:26 +0800
Subject: [PATCH] [Feature][Connector-V2][FlinkTableStore] Add split enumerator

---
 .../fts/config/FlinkTableStoreConfig.java     |  6 ++
 .../FlinkTableStoreSourceSplitEnumerator.java | 57 +++++++++++++++++++
 2 files changed, 63 insertions(+)
 create mode 100644 seatunnel-connectors-v2/connector-flink-table-store/src/main/java/org/apache/seatunnel/connectors/seatunnel/fts/source/FlinkTableStoreSourceSplitEnumerator.java

diff --git a/seatunnel-connectors-v2/connector-flink-table-store/src/main/java/org/apache/seatunnel/connectors/seatunnel/fts/config/FlinkTableStoreConfig.java b/seatunnel-connectors-v2/connector-flink-table-store/src/main/java/org/apache/seatunnel/connectors/seatunnel/fts/config/FlinkTableStoreConfig.java
index 2ec4f995f266..301319da0148 100644
--- a/seatunnel-connectors-v2/connector-flink-table-store/src/main/java/org/apache/seatunnel/connectors/seatunnel/fts/config/FlinkTableStoreConfig.java
+++ b/seatunnel-connectors-v2/connector-flink-table-store/src/main/java/org/apache/seatunnel/connectors/seatunnel/fts/config/FlinkTableStoreConfig.java
@@ -35,4 +35,10 @@ public class FlinkTableStoreConfig {
                     .listType()
                     .noDefaultValue()
                     .withDescription("The read columns of the flink table store");
+
+    public static final Option<String> HDFS_SITE_PATH =
+            Options.key("hdfs_site_path")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The file path of hdfs-site.xml");
 }
diff --git a/seatunnel-connectors-v2/connector-flink-table-store/src/main/java/org/apache/seatunnel/connectors/seatunnel/fts/source/FlinkTableStoreSourceSplitEnumerator.java b/seatunnel-connectors-v2/connector-flink-table-store/src/main/java/org/apache/seatunnel/connectors/seatunnel/fts/source/FlinkTableStoreSourceSplitEnumerator.java
new file mode 100644
index 000000000000..2c2e975282f6
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-flink-table-store/src/main/java/org/apache/seatunnel/connectors/seatunnel/fts/source/FlinkTableStoreSourceSplitEnumerator.java
@@ -0,0 +1,57 @@
+/*
+ * 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.fts.source;
+
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+
+import java.io.IOException;
+import java.util.List;
+
+public class FlinkTableStoreSourceSplitEnumerator
+        implements SourceSplitEnumerator<FlinkTableStoreSourceSplit, FlinkTableStoreSourceState> {
+    @Override
+    public void open() {}
+
+    @Override
+    public void run() throws Exception {}
+
+    @Override
+    public void close() throws IOException {}
+
+    @Override
+    public void addSplitsBack(List<FlinkTableStoreSourceSplit> splits, int subtaskId) {}
+
+    @Override
+    public int currentUnassignedSplitSize() {
+        return 0;
+    }
+
+    @Override
+    public void handleSplitRequest(int subtaskId) {}
+
+    @Override
+    public void registerReader(int subtaskId) {}
+
+    @Override
+    public FlinkTableStoreSourceState snapshotState(long checkpointId) throws Exception {
+        return null;
+    }
+
+    @Override
+    public void notifyCheckpointComplete(long checkpointId) throws Exception {}
+}