[Improve][Starrocks] Catch lable already exception

Author: hailin0

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManager.java

@@ -20,6 +20,7 @@
 import org.apache.seatunnel.shade.com.google.common.base.Strings;
 
 import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.common.utils.ExceptionUtils;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.config.SinkConfig;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.exception.StarRocksConnectorErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.exception.StarRocksConnectorException;
@@ -45,9 +46,16 @@ public class StarRocksSinkManager {
     private long batchBytesSize = 0;
 
     public StarRocksSinkManager(SinkConfig sinkConfig, TableSchema tableSchema) {
+        this(sinkConfig, tableSchema, new StarRocksStreamLoadVisitor(sinkConfig, tableSchema));
+    }
+
+    StarRocksSinkManager(
+            SinkConfig sinkConfig,
+            TableSchema tableSchema,
+            StarRocksStreamLoadVisitor streamLoadVisitor) {
         this.sinkConfig = sinkConfig;
         this.batchList = new ArrayList<>();
-        starrocksStreamLoadVisitor = new StarRocksStreamLoadVisitor(sinkConfig, tableSchema);
+        starrocksStreamLoadVisitor = streamLoadVisitor;
     }
 
     private void tryInit() throws IOException {
@@ -90,6 +98,13 @@ public synchronized void flush() throws IOException {
                 }
             } catch (Exception e) {
                 log.warn("Writing records to StarRocks failed, retry times = {}", i, e);
+
+                String labelAlreadyMessage =
+                        String.format("Label [%s] has already been used", label);
+                if (ExceptionUtils.getMessage(e).contains(labelAlreadyMessage)) {
+                    log.warn("Label [{}] has already been used, Skipping this batch", label);
+                    break;
+                }
                 if (i >= sinkConfig.getMaxRetries()) {
                     throw new StarRocksConnectorException(
                             StarRocksConnectorErrorCode.WRITE_RECORDS_FAILED,

seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManagerTest.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.starrocks.client;
+
+import org.apache.seatunnel.connectors.seatunnel.starrocks.config.SinkConfig;
+import org.apache.seatunnel.connectors.seatunnel.starrocks.exception.StarRocksConnectorException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class StarRocksSinkManagerTest {
+
+    private SinkConfig mockSinkConfig;
+    private StarRocksStreamLoadVisitor mockStreamLoadVisitor;
+    private StarRocksSinkManager sinkManager;
+
+    @BeforeEach
+    void setUp() {
+        mockSinkConfig = mock(SinkConfig.class);
+        mockStreamLoadVisitor = mock(StarRocksStreamLoadVisitor.class);
+        when(mockSinkConfig.getBatchMaxSize()).thenReturn(10);
+        when(mockSinkConfig.getMaxRetries()).thenReturn(3);
+        when(mockSinkConfig.getRetryBackoffMultiplierMs()).thenReturn(100);
+        when(mockSinkConfig.getMaxRetryBackoffMs()).thenReturn(1000);
+        this.sinkManager =
+                new StarRocksSinkManager(mockSinkConfig, null, mockStreamLoadVisitor) {
+                    public String createBatchLabel() {
+                        return "test-label";
+                    }
+                };
+    }
+
+    @Test
+    void testLabelAlreadyMessageHandledCorrectly() throws Exception {
+        // Mock behavior for label already used
+        doThrow(new RuntimeException("Label [test-label] has already been used"))
+                .when(mockStreamLoadVisitor)
+                .doStreamLoad(any());
+
+        // Add a record to trigger flush
+        sinkManager.write("test-record");
+
+        // Verify that the exception is caught and the batch is skipped
+        assertDoesNotThrow(() -> sinkManager.flush());
+        verify(mockStreamLoadVisitor, times(1)).doStreamLoad(any());
+    }
+
+    @Test
+    void testLabelAlreadyMessageNotHandled() throws Exception {
+        // Mock behavior for a different exception
+        doThrow(new RuntimeException("Some other error"))
+                .when(mockStreamLoadVisitor)
+                .doStreamLoad(any());
+
+        // Add a record to trigger flush
+        sinkManager.write("test-record");
+
+        // Verify that the exception is propagated after retries
+        assertThrows(StarRocksConnectorException.class, () -> sinkManager.flush());
+        verify(mockStreamLoadVisitor, times(4))
+                .doStreamLoad(any()); // 3 retries + 1 initial attempt
+    }
+}