[Improve] Improve StarRocks Sink Config (#4212)

Author: Hisoka-X

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/SinkConfig.java

@@ -64,9 +64,7 @@ public static SinkConfig of(ReadonlyConfig config) {
         SinkConfig sinkConfig = new SinkConfig();
         sinkConfig.setNodeUrls(config.get(StarRocksSinkOptions.NODE_URLS));
         sinkConfig.setDatabase(config.get(StarRocksSinkOptions.DATABASE));
-        sinkConfig.setJdbcUrl(String.format("jdbc:mysql://%s:%d",
-            sinkConfig.getNodeUrls().get(0).split(":")[0],
-            config.get(StarRocksSinkOptions.QUERY_PORT)));
+        sinkConfig.setJdbcUrl(config.get(StarRocksOptions.BASE_URL));
         config.getOptional(StarRocksOptions.USERNAME).ifPresent(sinkConfig::setUsername);
         config.getOptional(StarRocksOptions.PASSWORD).ifPresent(sinkConfig::setPassword);
         config.getOptional(StarRocksSinkOptions.TABLE).ifPresent(sinkConfig::setTable);

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/StarRocksSinkOptions.java

@@ -29,7 +29,7 @@ public interface StarRocksSinkOptions {
     Option<List<String>> NODE_URLS = Options.key("nodeUrls")
         .listType()
         .noDefaultValue()
-        .withDescription("StarRocks cluster address, the format is [\"fe_ip:fe_http_port\", ...]");
+        .withDescription("StarRocks cluster http address, the format is [\"fe_ip:fe_http_port\", ...]");
 
     Option<String> LABEL_PREFIX = Options.key("labelPrefix")
         .stringType()
@@ -58,11 +58,6 @@ public interface StarRocksSinkOptions {
             "    \"replication_num\" = \"1\" \n" +
             ")").withDescription("Create table statement template, used to create StarRocks table");
 
-    Option<Integer> QUERY_PORT = Options.key("query_port")
-        .intType()
-        .defaultValue(9030)
-        .withDescription("FE MySQL server port");
-
     Option<Integer> BATCH_MAX_SIZE = Options.key("batch_max_rows")
         .intType()
         .defaultValue(1024)

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkFactory.java

@@ -42,7 +42,7 @@ public String factoryIdentifier() {
     public OptionRule optionRule() {
         return OptionRule.builder()
             .required(StarRocksOptions.USERNAME, StarRocksOptions.PASSWORD)
-            .required(StarRocksSinkOptions.DATABASE, StarRocksSinkOptions.QUERY_PORT)
+            .required(StarRocksSinkOptions.DATABASE, StarRocksOptions.BASE_URL)
             .required(StarRocksSinkOptions.NODE_URLS)
             .optional(StarRocksSinkOptions.TABLE, StarRocksSinkOptions.LABEL_PREFIX, StarRocksSinkOptions.BATCH_MAX_SIZE, StarRocksSinkOptions.BATCH_MAX_BYTES,
                 StarRocksSinkOptions.BATCH_INTERVAL_MS, StarRocksSinkOptions.MAX_RETRIES, StarRocksSinkOptions.MAX_RETRY_BACKOFF_MS,

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSplitEnumeratorTask.java

@@ -132,15 +132,16 @@ public void triggerBarrier(Barrier barrier) throws Exception {
             this.prepareCloseBarrierId.set(barrier.getId());
         }
         final long barrierId = barrier.getId();
-        Serializable snapshotState = null;
+        Serializable snapshotState;
+        byte[] serialize = null;
         synchronized (enumeratorContext) {
             if (barrier.snapshot()) {
                 snapshotState = enumerator.snapshotState(barrierId);
+                serialize = enumeratorStateSerializer.serialize(snapshotState);
             }
             sendToAllReader(location -> new BarrierFlowOperation(barrier, location));
         }
         if (barrier.snapshot()) {
-            byte[] serialize = enumeratorStateSerializer.serialize(snapshotState);
             this.getExecutionContext().sendToMaster(new TaskAcknowledgeOperation(this.taskLocation, (CheckpointBarrier) barrier,
                 Collections.singletonList(new ActionSubtaskState(source.getId(), -1, Collections.singletonList(serialize)))));
         }