[hotfix-35505][kudu]

1.add [mutationBufferMaxOps];
2.if catch exception with "MANUAL_FLUSH is enabled but the buffer is too big", then session flush;

Author: FlechazoW

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.sink.kudu;
 
+import com.dtstack.flink.sql.exception.ExceptionTrace;
 import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
 import com.dtstack.flink.sql.sink.kudu.table.KuduTableInfo;
@@ -60,6 +61,8 @@ public class KuduOutputFormat extends AbstractDtRichOutputFormat<Tuple2<Boolean,
     private static final long serialVersionUID = 1L;
 
     private static final Logger LOG = LoggerFactory.getLogger(KuduOutputFormat.class);
+    private static final String MANUAL_FLUSH_BUFFER_BIG_MSG = "MANUAL_FLUSH is enabled but the buffer is too big";
+
     protected String[] fieldNames;
     TypeInformation<?>[] fieldTypes;
     boolean enableKrb;
@@ -76,6 +79,8 @@ public class KuduOutputFormat extends AbstractDtRichOutputFormat<Tuple2<Boolean,
 
     private Integer defaultOperationTimeoutMs;
 
+    private int mutationBufferMaxOps;
+
     /**
      * kerberos
      */
@@ -186,9 +191,7 @@ private KuduSession buildSessionWithFlushMode(String flushMode, KuduClient kuduC
         KuduSession kuduSession = kuduClient.newSession();
         if (flushMode.equalsIgnoreCase(KuduTableInfo.KuduFlushMode.MANUAL_FLUSH.name())) {
             kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
-            kuduSession.setMutationBufferSpace(
-                    Integer.parseInt(String.valueOf(Math.round(batchSize * 1.2)))
-            );
+            kuduSession.setMutationBufferSpace(mutationBufferMaxOps);
         }
 
         if (flushMode.equalsIgnoreCase(KuduTableInfo.KuduFlushMode.AUTO_FLUSH_SYNC.name())) {
@@ -197,8 +200,9 @@ private KuduSession buildSessionWithFlushMode(String flushMode, KuduClient kuduC
         }
 
         if (flushMode.equalsIgnoreCase(KuduTableInfo.KuduFlushMode.AUTO_FLUSH_BACKGROUND.name())) {
-            LOG.warn("Unable to determine the order of data at AUTO_FLUSH_BACKGROUND mode.");
+            LOG.warn("Unable to determine the order of data at AUTO_FLUSH_BACKGROUND mode. Only [batchWaitInterval] will effect.");
             kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
+            kuduSession.setFlushInterval(batchWaitInterval);
         }
 
         return kuduSession;
@@ -213,21 +217,26 @@ public void writeRecord(Tuple2<Boolean, Row> record) throws IOException {
         Row row = record.getField(1);
 
         try {
-            if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0) {
-                LOG.info("Receive data : {}", row);
-            }
             if (rowCount.getAndIncrement() >= batchSize) {
                 flush();
             }
             // At AUTO_FLUSH_SYNC mode, kudu automatically flush once session apply operation, then get the response from kudu server.
             if (flushMode.equalsIgnoreCase(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC.name())) {
                 dealResponse(session.apply(toOperation(writeMode, row)));
+            } else {
+                session.apply(toOperation(writeMode, row));
             }
-
-            session.apply(toOperation(writeMode, row));
             outRecords.inc();
+            if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0) {
+                LOG.info("Receive data : {}", row);
+            }
         } catch (KuduException e) {
-            throw new RuntimeException(e);
+            // 如果出现了buffer is too big的问题，需要将当前buffer里的数据flush掉即可。
+            if (ExceptionTrace.traceOriginalCause(e).contains(MANUAL_FLUSH_BUFFER_BIG_MSG)) {
+                flush();
+            } else {
+                throw new RuntimeException(e);
+            }
         }
     }
 
@@ -483,6 +492,11 @@ public KuduOutputFormatBuilder setFlushMode(String flushMode) {
             return this;
         }
 
+        public KuduOutputFormatBuilder setMutationBufferMaxOps(Integer mutationBufferMaxOps) {
+            kuduOutputFormat.mutationBufferMaxOps = mutationBufferMaxOps;
+            return this;
+        }
+
         public KuduOutputFormat finish() {
             if (kuduOutputFormat.kuduMasters == null) {
                 throw new IllegalArgumentException("No kuduMasters supplied.");

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduSink.java

@@ -36,6 +36,7 @@ public class KuduSink implements RetractStreamTableSink<Row>, Serializable, IStr
     private Integer batchSize;
     private Integer batchWaitInterval;
     private String flushMode;
+    private Integer mutationBufferMaxOps;
 
     @Override
     public KuduSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
@@ -55,6 +56,9 @@ public KuduSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         this.batchSize = kuduTableInfo.getBatchSize();
         this.batchWaitInterval = kuduTableInfo.getBatchWaitInterval();
         this.flushMode = kuduTableInfo.getFlushMode();
+        this.mutationBufferMaxOps = Objects.isNull(kuduTableInfo.getMutationBufferMaxOps()) ?
+            Integer.parseInt(String.valueOf(Math.round(batchSize * 1.2))) :
+            kuduTableInfo.getMutationBufferMaxOps();
         return this;
     }
 
@@ -80,6 +84,7 @@ public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<
                 .setBatchSize(this.batchSize)
                 .setBatchWaitInterval(this.batchWaitInterval)
                 .setFlushMode(this.flushMode)
+                .setMutationBufferMaxOps(this.mutationBufferMaxOps)
                 .finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(kuduOutputFormat);
         DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction);

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/table/KuduSinkParser.java

@@ -32,6 +32,8 @@ public class KuduSinkParser extends AbstractTableParser {
 
     public static final String BATCH_WAIT_INTERVAL_KEY = "batchWaitInterval";
 
+    public static final String BUFFER_MAX_KEY = "mutationBufferMaxOps";
+
     public static final Integer DEFAULT_BATCH_WAIT_INTERVAL = 60 * 1000;
 
     public static final String SESSION_FLUSH_MODE_KEY = "flushMode";
@@ -50,6 +52,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kuduTableInfo.setDefaultOperationTimeoutMs(MathUtil.getIntegerVal(props.get(OPERATION_TIMEOUT_MS.toLowerCase())));
         kuduTableInfo.setBatchSize(MathUtil.getIntegerVal(props.getOrDefault(BATCH_SIZE_KEY.toLowerCase(), DEFAULT_BATCH_SIZE)));
         kuduTableInfo.setBatchWaitInterval(MathUtil.getIntegerVal(props.getOrDefault(BATCH_WAIT_INTERVAL_KEY.toLowerCase(), DEFAULT_BATCH_WAIT_INTERVAL)));
+        kuduTableInfo.setMutationBufferMaxOps(MathUtil.getIntegerVal(props.get(BUFFER_MAX_KEY.toLowerCase())));
 
         if (Objects.isNull(props.get(SESSION_FLUSH_MODE_KEY.toLowerCase()))) {
             if (kuduTableInfo.getBatchSize() > 1) {

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/table/KuduTableInfo.java

@@ -19,6 +19,8 @@ public class KuduTableInfo extends AbstractTargetTableInfo implements KerberosTa
 
     private Integer defaultOperationTimeoutMs;
 
+    private Integer mutationBufferMaxOps;
+
     /**
      * kerberos
      */
@@ -80,6 +82,14 @@ public void setDefaultOperationTimeoutMs(Integer defaultOperationTimeoutMs) {
         this.defaultOperationTimeoutMs = defaultOperationTimeoutMs;
     }
 
+    public Integer getMutationBufferMaxOps() {
+        return mutationBufferMaxOps;
+    }
+
+    public void setMutationBufferMaxOps(Integer mutationBufferMaxOps) {
+        this.mutationBufferMaxOps = mutationBufferMaxOps;
+    }
+
     @Override
     public boolean check() {
         Preconditions.checkNotNull(kuduMasters, "kudu field of kuduMasters is required");