add check for doris.batch.size and improve some log print

JNSimba · Sep 3, 2024 · 0f1615b · 0f1615b
1 parent 1f9f12d
commit 0f1615b
Show file tree

Hide file tree

Showing 5 changed files with 9 additions and 3 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java
@@ -42,7 +42,7 @@ public interface ConfigurationOptions {
     Integer DORIS_TABLET_SIZE_MIN = 1;
 
     String DORIS_BATCH_SIZE = "doris.batch.size";
-    Integer DORIS_BATCH_SIZE_DEFAULT = 1024;
+    Integer DORIS_BATCH_SIZE_DEFAULT = 4064;
 
     String DORIS_EXEC_MEM_LIMIT = "doris.exec.mem.limit";
     Long DORIS_EXEC_MEM_LIMIT_DEFAULT = 8589934592L;

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java
@@ -17,6 +17,8 @@
 
 package org.apache.doris.flink.cfg;
 
+import org.apache.flink.util.Preconditions;
+
 import java.io.Serializable;
 import java.util.Objects;
 
@@ -276,6 +278,9 @@ public Builder setFlightSqlPort(Integer flightSqlPort) {
         }
 
         public DorisReadOptions build() {
+            Preconditions.checkArgument(
+                    requestBatchSize >= 1 && requestBatchSize <= 65535,
+                    "batchSize should be between 1 and 65535");
             return new DorisReadOptions(
                     readFields,
                     filterQuery,

diff --git a/...pache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImplV2.java b/...pache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChangeImplV2.java
@@ -123,7 +123,7 @@ public boolean schemaChange(JsonNode recordRoot) {
                     String dorisTbl = getCreateTableIdentifier(recordRoot);
                     changeContext.getTableMapping().put(cdcTbl, dorisTbl);
                     this.tableMapping = changeContext.getTableMapping();
-                    LOG.info("create table ddl status: {}", status);
+                    LOG.info("create table ddl status: {}, add tableMapping {},{}", status, cdcTbl, dorisTbl);
                 }
             } else if (eventType.equals(EventType.ALTER)) {
                 Tuple2<String, String> dorisTableTuple = getDorisTableTuple(recordRoot);

diff --git a/...ava/org/apache/doris/flink/sink/writer/serializer/jsondebezium/SQLParserSchemaChange.java b/...ava/org/apache/doris/flink/sink/writer/serializer/jsondebezium/SQLParserSchemaChange.java
@@ -83,7 +83,7 @@ public boolean schemaChange(JsonNode recordRoot) {
                     String dorisTbl = getCreateTableIdentifier(recordRoot);
                     changeContext.getTableMapping().put(cdcTbl, dorisTbl);
                     this.tableMapping = changeContext.getTableMapping();
-                    LOG.info("create table ddl status: {}", status);
+                    LOG.info("create table ddl status: {}, add tableMapping {},{}", status, cdcTbl, dorisTbl);
                 }
             } else if (eventType.equals(EventType.ALTER)) {
                 Tuple2<String, String> dorisTableTuple = getDorisTableTuple(recordRoot);

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java
@@ -158,6 +158,7 @@ public void build() throws Exception {
             System.out.println("Create table finished.");
             System.exit(0);
         }
+        LOG.info("table mapping: {}", tableMapping);
         config.setString(TABLE_NAME_OPTIONS, getSyncTableList(syncTables));
         DataStreamSource<String> streamSource = buildCdcSource(env);
         if (singleSink) {