From 9fe85caf7972080f7b7b66e0efee3b878a2da1be Mon Sep 17 00:00:00 2001
From: wudi <676366545@qq.com>
Date: Thu, 9 May 2024 16:09:49 +0800
Subject: [PATCH] multiple concurrency, make streamload  on each BE are more
 balanced.

---
 .../java/org/apache/doris/flink/sink/writer/DorisWriter.java  | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java
index 9e8c30f29..6930534ba 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java
@@ -66,6 +66,7 @@ public class DorisWriter<IN> implements SinkWriter<IN, DorisCommittable, DorisWr
     private final DorisReadOptions dorisReadOptions;
     private final DorisExecutionOptions executionOptions;
     private final String labelPrefix;
+    private final int subtaskId;
     private final LabelGenerator labelGenerator;
     private final int intervalTime;
     private final DorisWriterState dorisWriterState;
@@ -91,6 +92,7 @@ public DorisWriter(Sink.InitContext initContext,
         LOG.info("labelPrefix " + executionOptions.getLabelPrefix());
         this.dorisWriterState = new DorisWriterState(executionOptions.getLabelPrefix());
         this.labelPrefix = executionOptions.getLabelPrefix() + "_" + initContext.getSubtaskId();
+        this.subtaskId = initContext.getSubtaskId();
         this.labelGenerator = new LabelGenerator(labelPrefix, executionOptions.enabled2PC());
         this.scheduledExecutorService = new ScheduledThreadPoolExecutor(1, new ExecutorThreadFactory("stream-load-check"));
         this.serializer = serializer;
@@ -240,7 +242,7 @@ public void close() throws Exception {
     public String getAvailableBackend() {
         long tmp = pos + backends.size();
         while (pos < tmp) {
-            BackendV2.BackendRowV2 backend = backends.get((int) (pos % backends.size()));
+            BackendV2.BackendRowV2 backend = backends.get((int) ((pos + subtaskId) % backends.size()));
             String res = backend.toBackendString();
             if(tryHttpConnection(res)){
                 pos++;