modify structure

Author: yanxi0227

mysql/mysql-sink/src/main/java/com/dtstack/flink/sql/sink/mysql/MysqlSink.java

@@ -22,6 +22,7 @@
 
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
+import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
 
 import java.util.List;
 import java.util.Map;
@@ -40,6 +41,11 @@ public class MysqlSink extends RdbSink implements IStreamSinkGener<RdbSink> {
     public MysqlSink() {
     }
 
+    @Override
+    public RetractJDBCOutputFormat getOutputFormat() {
+        return new RetractJDBCOutputFormat();
+    }
+
     @Override
     public void buildSql(String tableName, List<String> fields) {
         buildInsertSql(tableName, fields);

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/RdbSink.java

@@ -17,11 +17,8 @@
  */
 package com.dtstack.flink.sql.sink.rdb;
 
-import com.dtstack.flink.sql.enums.EDatabaseType;
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
-import com.dtstack.flink.sql.sink.rdb.format.OracleOutputFormat;
 import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
-import com.dtstack.flink.sql.sink.rdb.format.SqlserverOutputFormat;
 import com.dtstack.flink.sql.sink.rdb.table.RdbTableInfo;
 import com.dtstack.flink.sql.table.TargetTableInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -85,7 +82,7 @@ public RichSinkFunction createJdbcSinkFunc() {
             throw new RuntimeException("any of params in(driverName, dbURL, userName, password, type, tableName) " +
                     " must not be null. please check it!!!");
         }
-        RetractJDBCOutputFormat outputFormat = getOutputFormat(dbType);
+        RetractJDBCOutputFormat outputFormat = getOutputFormat();
         outputFormat.setDbURL(dbURL);
         outputFormat.setDrivername(driverName);
         outputFormat.setUsername(userName);
@@ -102,15 +99,7 @@ public RichSinkFunction createJdbcSinkFunc() {
         return outputFormatSinkFunc;
     }
 
-    protected RetractJDBCOutputFormat getOutputFormat(String dbType) {
-        if (dbType.equalsIgnoreCase(EDatabaseType.SQLSERVER.name())) {
-            return new SqlserverOutputFormat();
-        } else if (dbType.equalsIgnoreCase(EDatabaseType.ORACLE.name())) {
-            return new OracleOutputFormat();
-        } else {
-            return new RetractJDBCOutputFormat();
-        }
-    }
+    public abstract RetractJDBCOutputFormat getOutputFormat();
 
 
     @Override

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/OracleOutputFormat.java

@@ -1,22 +1,24 @@
-/**
+/*
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *     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.dtstack.flink.sql.sink.rdb.format;
 
+package com.dtstack.flink.sql.sink.sqlserver;
+
+import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
 import java.sql.ResultSet;

…nk/rdb/format/SqlserverOutputFormat.java …ink/sqlserver/SqlserverOutputFormat.javardb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/SqlserverOutputFormat.java renamed to sqlserver/sqlserver-sink/src/main/java/com/dtstack/flink/sql/sink/sqlserver/SqlserverOutputFormat.java rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/SqlserverOutputFormat.java renamed to sqlserver/sqlserver-sink/src/main/java/com/dtstack/flink/sql/sink/sqlserver/SqlserverOutputFormat.java

@@ -19,6 +19,7 @@
 
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
+import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.*;
@@ -38,6 +39,11 @@ public String getDriverName() {
         return SQLSERVER_DRIVER;
     }
 
+    @Override
+    public RetractJDBCOutputFormat getOutputFormat() {
+        return new SqlserverOutputFormat();
+    }
+
     @Override
     public void buildSql(String tableName, List<String> fields) {
         buildInsertSql(tableName, fields);