This is an automated email from the ASF dual-hosted git repository.
corgy pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new b6fc222c0a [Improve][Starrocks] Catch lable already exception (#9222)
b6fc222c0a is described below
commit b6fc222c0adbda886a635a6f0577302951d548e3
Author: hailin0 <[email protected]>
AuthorDate: Wed Apr 30 09:47:37 2025 +0800
[Improve][Starrocks] Catch lable already exception (#9222)
---
.../starrocks/client/StarRocksSinkManager.java | 17 ++++-
.../starrocks/client/StarRocksSinkManagerTest.java | 88 ++++++++++++++++++++++
2 files changed, 104 insertions(+), 1 deletion(-)
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManager.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManager.java
index 9a87169884..5fcbf59922 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManager.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManager.java
@@ -20,6 +20,7 @@ package
org.apache.seatunnel.connectors.seatunnel.starrocks.client;
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 class StarRocksSinkManager {
}
} 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,
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManagerTest.java
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManagerTest.java
new file mode 100644
index 0000000000..4cebd23dd1
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManagerTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.getBatchMaxBytes()).thenReturn(1024 * 1024 *
1024L);
+ 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
+ }
+}
