[Improve][Core] Move MultiTableSink to seatunnel-api module (apache#7243

) * [Improve][Core] Move MultiTableSink to seatunnel-api module * [Improve][Core] Move MultiTableSink to seatunnel-api module
jovezhong · Jul 23, 2024 · cc59499 · cc59499
1 parent fe0c477
commit cc59499
Show file tree

Hide file tree

Showing 13 changed files with 19 additions and 21 deletions.
diff --git a/...esink/MultiTableAggregatedCommitInfo.java → ...esink/MultiTableAggregatedCommitInfo.java b/...esink/MultiTableAggregatedCommitInfo.java → ...esink/MultiTableAggregatedCommitInfo.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;

diff --git a/.../multitablesink/MultiTableCommitInfo.java → .../multitablesink/MultiTableCommitInfo.java b/.../multitablesink/MultiTableCommitInfo.java → .../multitablesink/MultiTableCommitInfo.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;

diff --git a/...common/multitablesink/MultiTableSink.java → ...i/sink/multitablesink/MultiTableSink.java b/...common/multitablesink/MultiTableSink.java → ...i/sink/multitablesink/MultiTableSink.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import org.apache.seatunnel.api.common.JobContext;
 import org.apache.seatunnel.api.serialization.DefaultSerializer;
@@ -28,6 +28,8 @@
 import org.apache.seatunnel.api.table.factory.MultiTableFactoryContext;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 
+import lombok.Getter;
+
 import java.io.IOException;
 import java.util.Collection;
 import java.util.HashMap;
@@ -44,7 +46,7 @@ public class MultiTableSink
                 MultiTableCommitInfo,
                 MultiTableAggregatedCommitInfo> {
 
-    private final Map<String, SeaTunnelSink> sinks;
+    @Getter private final Map<String, SeaTunnelSink> sinks;
     private final int replicaNum;
 
     public MultiTableSink(MultiTableFactoryContext context) {

diff --git a/...nk/MultiTableSinkAggregatedCommitter.java → ...nk/MultiTableSinkAggregatedCommitter.java b/...nk/MultiTableSinkAggregatedCommitter.java → ...nk/MultiTableSinkAggregatedCommitter.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import org.apache.seatunnel.api.sink.MultiTableResourceManager;
 import org.apache.seatunnel.api.sink.SinkAggregatedCommitter;

diff --git a/...ltitablesink/MultiTableSinkCommitter.java → ...ltitablesink/MultiTableSinkCommitter.java b/...ltitablesink/MultiTableSinkCommitter.java → ...ltitablesink/MultiTableSinkCommitter.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import org.apache.seatunnel.api.sink.SinkCommitter;
 

diff --git a/...multitablesink/MultiTableSinkFactory.java → ...multitablesink/MultiTableSinkFactory.java b/...multitablesink/MultiTableSinkFactory.java → ...multitablesink/MultiTableSinkFactory.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.connector.TableSink;

diff --git a/.../multitablesink/MultiTableSinkWriter.java → .../multitablesink/MultiTableSinkWriter.java b/.../multitablesink/MultiTableSinkWriter.java → .../multitablesink/MultiTableSinkWriter.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import org.apache.seatunnel.api.sink.MultiTableResourceManager;
 import org.apache.seatunnel.api.sink.SinkWriter;

diff --git a/...ommon/multitablesink/MultiTableState.java → .../sink/multitablesink/MultiTableState.java b/...ommon/multitablesink/MultiTableState.java → .../sink/multitablesink/MultiTableState.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;

diff --git a/...titablesink/MultiTableWriterRunnable.java → ...titablesink/MultiTableWriterRunnable.java b/...titablesink/MultiTableWriterRunnable.java → ...titablesink/MultiTableWriterRunnable.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;

diff --git a/...mmon/multitablesink/SinkContextProxy.java → ...sink/multitablesink/SinkContextProxy.java b/...mmon/multitablesink/SinkContextProxy.java → ...sink/multitablesink/SinkContextProxy.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import org.apache.seatunnel.api.common.metrics.MetricsContext;
 import org.apache.seatunnel.api.event.EventListener;

diff --git a/...common/multitablesink/SinkIdentifier.java → ...i/sink/multitablesink/SinkIdentifier.java b/...common/multitablesink/SinkIdentifier.java → ...i/sink/multitablesink/SinkIdentifier.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.common.multitablesink;
+package org.apache.seatunnel.api.sink.multitablesink;
 
 import lombok.EqualsAndHashCode;
 import lombok.Getter;

diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/FactoryUtil.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/FactoryUtil.java
@@ -23,6 +23,7 @@
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.env.ParsingMode;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.multitablesink.MultiTableSinkFactory;
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.source.SourceOptions;
 import org.apache.seatunnel.api.source.SourceSplit;
@@ -151,7 +152,7 @@ SeaTunnelSink<IN, StateT, CommitInfoT, AggregatedCommitInfoT> createMultiTableSi
                     ClassLoader classLoader) {
         try {
             TableSinkFactory<IN, StateT, CommitInfoT, AggregatedCommitInfoT> factory =
-                    discoverFactory(classLoader, TableSinkFactory.class, "MultiTableSink");
+                    new MultiTableSinkFactory();
             MultiTableFactoryContext context =
                     new MultiTableFactoryContext(options, classLoader, sinks);
             ConfigValidator.of(context.getOptions()).validate(factory.optionRule());

diff --git a/...nnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java b/...nnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java
@@ -26,9 +26,9 @@
 import org.apache.seatunnel.api.sink.SaveModeHandler;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.sink.SupportSaveMode;
+import org.apache.seatunnel.api.sink.multitablesink.MultiTableSink;
 import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
 import org.apache.seatunnel.common.utils.ExceptionUtils;
-import org.apache.seatunnel.common.utils.ReflectionUtils;
 import org.apache.seatunnel.common.utils.RetryUtils;
 import org.apache.seatunnel.common.utils.SeaTunnelException;
 import org.apache.seatunnel.engine.checkpoint.storage.exception.CheckpointStorageException;
@@ -375,13 +375,8 @@ public static void handleSaveMode(SeaTunnelSink sink) {
                     throw new SeaTunnelRuntimeException(HANDLE_SAVE_MODE_FAILED, e);
                 }
             }
-        } else if (sink.getClass()
-                .getName()
-                .equals(
-                        "org.apache.seatunnel.connectors.seatunnel.common.multitablesink.MultiTableSink")) {
-            // TODO we should not use class name to judge the sink type
-            Map<String, SeaTunnelSink> sinks =
-                    (Map<String, SeaTunnelSink>) ReflectionUtils.getField(sink, "sinks").get();
+        } else if (sink instanceof MultiTableSink) {
+            Map<String, SeaTunnelSink> sinks = ((MultiTableSink) sink).getSinks();
             for (SeaTunnelSink seaTunnelSink : sinks.values()) {
                 handleSaveMode(seaTunnelSink);
             }