[pipeline][starrocks] support sink.buffer-flush.max-rows

Author: guoxuanlin

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksDataSinkFactory.java

@@ -97,6 +97,10 @@ private StarRocksSinkOptions buildSinkConnectorOptions(Configuration cdcConfig)
                 .getOptional(StarRocksDataSinkOptions.SINK_BATCH_MAX_SIZE)
                 .ifPresent(
                         config -> sinkConfig.set(StarRocksSinkOptions.SINK_BATCH_MAX_SIZE, config));
+        cdcConfig
+                .getOptional(StarRocksDataSinkOptions.SINK_BATCH_MAX_ROWS)
+                .ifPresent(
+                        config -> sinkConfig.set(StarRocksSinkOptions.SINK_BATCH_MAX_ROWS, config));
         cdcConfig
                 .getOptional(StarRocksDataSinkOptions.SINK_BATCH_FLUSH_INTERVAL)
                 .ifPresent(
@@ -175,6 +179,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         optionalOptions.add(StarRocksDataSinkOptions.SINK_SOCKET_TIMEOUT);
         optionalOptions.add(StarRocksDataSinkOptions.SINK_WAIT_FOR_CONTINUE_TIMEOUT);
         optionalOptions.add(StarRocksDataSinkOptions.SINK_BATCH_MAX_SIZE);
+        optionalOptions.add(StarRocksDataSinkOptions.SINK_BATCH_MAX_ROWS);
         optionalOptions.add(StarRocksDataSinkOptions.SINK_BATCH_FLUSH_INTERVAL);
         optionalOptions.add(StarRocksDataSinkOptions.SINK_SCAN_FREQUENCY);
         optionalOptions.add(StarRocksDataSinkOptions.SINK_IO_THREAD_COUNT);

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksDataSinkOptions.java

@@ -94,6 +94,12 @@ public class StarRocksDataSinkOptions {
                     .defaultValue(150L * 1024 * 1024)
                     .withDescription("Max data bytes of the flush.");
 
+    public static final ConfigOption<Long> SINK_BATCH_MAX_ROWS =
+            ConfigOptions.key("sink.buffer-flush.max-rows")
+                    .longType()
+                    .defaultValue(500000L)
+                    .withDescription("Max row count of the flush.");
+
     public static final ConfigOption<Long> SINK_BATCH_FLUSH_INTERVAL =
             ConfigOptions.key("sink.buffer-flush.interval-ms")
                     .longType()