fix conflict

Author: todd5167

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java

@@ -69,7 +69,7 @@
  * @see Tuple
  * @see DriverManager
  */
-public class CassandraOutputFormat extends DtRichOutputFormat {
+public class CassandraOutputFormat extends DtRichOutputFormat<Tuple2> {
     private static final long serialVersionUID = -7994311331389155692L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);

console/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/ConsoleOutputFormat.java

@@ -37,7 +37,7 @@
  *
  * @author xuqianjin
  */
-public class ConsoleOutputFormat extends DtRichOutputFormat {
+public class ConsoleOutputFormat extends DtRichOutputFormat<Tuple2> {
 
     private static final Logger LOG = LoggerFactory.getLogger(ConsoleOutputFormat.class);
 

core/src/main/java/com/dtstack/flink/sql/outputformat/DtRichOutputFormat.java

@@ -29,9 +29,9 @@
  */
 public abstract class DtRichOutputFormat<T> extends RichOutputFormat<T>{
 
-    protected transient Counter outRecords;
-    protected transient Counter outDirtyRecords;
-    protected transient Meter outRecordsRate;
+    public transient Counter outRecords;
+    public transient Counter outDirtyRecords;
+    public transient Meter outRecordsRate;
 
     protected static int ROW_PRINT_FREQUENCY = 1000;
     protected static int DIRTY_PRINT_FREQUENCY = 1000;

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

@@ -18,9 +18,8 @@
 
 package com.dtstack.flink.sql.sink.rdb.format;
 
-import com.dtstack.flink.sql.sink.MetricOutputFormat;
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
 import com.dtstack.flink.sql.util.JDBCUtils;
-import org.apache.flink.api.common.io.RichOutputFormat;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
@@ -37,7 +36,7 @@
  * @see Row
  * @see DriverManager
  */
-public abstract class AbstractJDBCOutputFormat<T> extends MetricOutputFormat<T> {
+public abstract class AbstractJDBCOutputFormat<T> extends DtRichOutputFormat<T> {
 
 	private static final long serialVersionUID = 1L;
     public static final int DEFAULT_FLUSH_MAX_SIZE = 100;

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

@@ -18,7 +18,7 @@
 
 package com.dtstack.flink.sql.sink.rdb.writer;
 
-import com.dtstack.flink.sql.sink.MetricOutputFormat;
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
@@ -47,9 +47,9 @@ public class AppendOnlyWriter implements JDBCWriter {
     private transient PreparedStatement statement;
     private transient List<Row> rows;
     // only use metric
-    private transient MetricOutputFormat metricOutputFormat;
+    private transient DtRichOutputFormat metricOutputFormat;
 
-    public AppendOnlyWriter(String insertSQL, int[] fieldTypes, MetricOutputFormat metricOutputFormat) {
+    public AppendOnlyWriter(String insertSQL, int[] fieldTypes, DtRichOutputFormat metricOutputFormat) {
         this.insertSQL = insertSQL;
         this.fieldTypes = fieldTypes;
         this.metricOutputFormat = metricOutputFormat;

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

@@ -18,7 +18,7 @@
 
 package com.dtstack.flink.sql.sink.rdb.writer;
 
-import com.dtstack.flink.sql.sink.MetricOutputFormat;
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
 import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.types.Row;
@@ -58,7 +58,7 @@ public static UpsertWriter create(
             String[] partitionFields,
             boolean objectReuse,
             boolean allReplace,
-            MetricOutputFormat metricOutputFormat) {
+            DtRichOutputFormat metricOutputFormat) {
 
         checkNotNull(keyFields);
 
@@ -96,9 +96,9 @@ public static UpsertWriter create(
     private transient Map<Row, Tuple2<Boolean, Row>> keyToRows;
     private transient PreparedStatement deleteStatement;
     // only use metric
-    private transient MetricOutputFormat metricOutputFormat;
+    private transient DtRichOutputFormat metricOutputFormat;
 
-    private UpsertWriter(int[] fieldTypes, int[] pkFields, int[] pkTypes, String deleteSQL, boolean objectReuse, MetricOutputFormat metricOutputFormat) {
+    private UpsertWriter(int[] fieldTypes, int[] pkFields, int[] pkTypes, String deleteSQL, boolean objectReuse, DtRichOutputFormat metricOutputFormat) {
         this.fieldTypes = fieldTypes;
         this.pkFields = pkFields;
         this.pkTypes = pkTypes;
@@ -216,7 +216,7 @@ private UpsertWriterUsingUpsertStatement(
                 boolean objectReuse,
                 String deleteSQL,
                 String upsertSQL,
-                MetricOutputFormat metricOutputFormat) {
+                DtRichOutputFormat metricOutputFormat) {
             super(fieldTypes, pkFields, pkTypes, deleteSQL, objectReuse, metricOutputFormat);
             this.upsertSQL = upsertSQL;
         }
@@ -274,7 +274,7 @@ private UpsertWriterUsingInsertUpdateStatement(
                 String existSQL,
                 String insertSQL,
                 String updateSQL,
-                MetricOutputFormat metricOutputFormat) {
+                DtRichOutputFormat metricOutputFormat) {
             super(fieldTypes, pkFields, pkTypes, deleteSQL, objectReuse, metricOutputFormat);
             this.existSQL = existSQL;
             this.insertSQL = insertSQL;