This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12692-ff586625703eac77000109ecb9e2a646af655e36 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 787380c35a2affb8a582f6abee0846a1e85a02ac Author: Jast <[email protected]> AuthorDate: Sat Oct 17 07:16:29 2026 +0000 [Fix][Connector-V2] Verify StarRocks label state for Publish Timeout before releasing batch (#12692) Co-authored-by: jast <[email protected]> Co-authored-by: Claude Code <[email protected]> --- docs/en/connectors/sink/StarRocks.md | 13 ++ docs/zh/connectors/sink/StarRocks.md | 13 ++ .../client/StarRocksStreamLoadVisitor.java | 124 +++++++++------- .../seatunnel/starrocks/config/SinkConfig.java | 3 + .../starrocks/config/StarRocksSinkOptions.java | 10 ++ .../starrocks/sink/StarRocksSinkFactory.java | 3 + .../client/StarRocksStreamLoadVisitorTest.java | 157 +++++++++++++++++++-- 7 files changed, 260 insertions(+), 63 deletions(-) diff --git a/docs/en/connectors/sink/StarRocks.md b/docs/en/connectors/sink/StarRocks.md index 1467424fd8..ba8c121a06 100644 --- a/docs/en/connectors/sink/StarRocks.md +++ b/docs/en/connectors/sink/StarRocks.md @@ -52,6 +52,7 @@ The internal implementation of StarRocks sink connector is cached and imported b | save_mode_create_template | string | no | see below | see below | | starrocks.config | map | no | - | The parameter of the stream load `data_desc` | | http_socket_timeout_ms | int | no | 180000 | Set http socket timeout, default is 3 minutes. | +| label_state_timeout_ms | long | no | 180000 | Total time the sink polls `get_load_state` to resolve a non-final stream load outcome (`Publish Timeout`, `Label Already Exists`, reused-label failure) to `VISIBLE`/`COMMITTED`/`ABORTED` before failing the flush. Default is 3 minutes; keep it below the job's checkpoint timeout. | | schema_save_mode | Enum | no | CREATE_SCHEMA_WHEN_NOT_EXIST | Before the synchronous task is turned on, different treatment schemes are selected for the existing surface structure of the target side. | | data_save_mode | Enum | no | APPEND_DATA | Before the synchronous task is turned on, different processing schemes are selected for data existing data on the target side. | | table_options | Map | no | - | Sink-specific table properties merged into CREATE TABLE PROPERTIES during SaveMode auto-create. See below. | @@ -170,6 +171,18 @@ sink { } ``` +### Non-final stream load outcomes + +A stream load response such as `Publish Timeout`, `Label Already Exists`, a reused-label failure, or an unrecognized status does not confirm the transaction outcome: a publish that timed out may still be aborted later (for example after a front-end restart). Instead of releasing the batch right away, the sink polls `get_load_state` for the batch label until the state is terminal: + +- `VISIBLE`/`COMMITTED`: the transaction is confirmed and the batch is released. +- `ABORTED`: the batch never became durable and is resent under a new label. +- anything else (`PREPARE`, `UNKNOWN`, unreadable response): polling continues until `label_state_timeout_ms` expires, then the flush fails and the job replays the batch from its last checkpoint. + +The poll runs synchronously on the checkpoint path, and the sink retry loop can repeat it up to `max_retries + 1` times, so the worst-case blocking time is roughly `(max_retries + 1) * label_state_timeout_ms`. Keep `label_state_timeout_ms` below the job's checkpoint timeout so an unresolved transaction surfaces as a checkpoint failure instead of holding the barrier for minutes. + +Because a failed flush is replayed from the checkpoint, batches that were already visible earlier in the same window can be submitted again. The sink remains at-least-once: DUPLICATE KEY tables may see duplicate rows after a replay, while Primary Key tables absorb them. + ### Zeta Timer Flush This engine-level capability is available only in Zeta. Configure `sink.flush.interval` in `env` to periodically write buffered rows through StarRocks Stream Load even when `batch_max_rows` and `batch_max_bytes` have not been reached. Spark and Flink do not trigger this scheduled flush. diff --git a/docs/zh/connectors/sink/StarRocks.md b/docs/zh/connectors/sink/StarRocks.md index 1cf935b7d3..166bad85cb 100644 --- a/docs/zh/connectors/sink/StarRocks.md +++ b/docs/zh/connectors/sink/StarRocks.md @@ -52,6 +52,7 @@ StarRocks数据接收器内部实现采用了缓存，通过stream load将数据 | save_mode_create_template | string | 否 | 参见表下方的说明 | 自动建表模板，详见表下方说明 | | starrocks.config | map | 否 | - | Stream Load `data_desc` 参数 | | http_socket_timeout_ms | int | 否 | 180000 | HTTP socket 超时时间，默认为 3 分钟 | +| label_state_timeout_ms | long | 否 | 180000 | 当 Stream Load 返回非终态结果（`Publish Timeout`、`Label Already Exists`、复用 label 的失败）时，接收器轮询 `get_load_state` 以确认事务状态的总时长，超时后刷新失败并由作业从 checkpoint 重放。默认 3 分钟，应小于作业的 checkpoint 超时时间。 | | schema_save_mode | Enum | 否 | CREATE_SCHEMA_WHEN_NOT_EXIST | 同步任务启动前，针对目标端已存在的表结构选择不同处理方式 | | data_save_mode | Enum | 否 | APPEND_DATA | 同步任务启动前，针对目标端已存在的数据选择不同处理方式 | | table_options | Map | 否 | - | SaveMode 自动建表时合并进 CREATE TABLE PROPERTIES 的 Sink 专属表属性，详见表下方说明 | @@ -165,6 +166,18 @@ sink { } ``` +### 非终态 Stream Load 结果 + +当 Stream Load 返回 `Publish Timeout`、`Label Already Exists`、复用 label 的失败或无法识别的状态时，事务结果尚未确认：发布超时的导入之后仍可能被中止（例如前端重启后）。接收器不会立即释放该批数据，而是持续查询该批次 label 的 `get_load_state`，直到状态变为终态： + +- `VISIBLE`/`COMMITTED`：事务已确认，释放该批数据。 +- `ABORTED`：该批数据未持久化，使用新 label 重新发送。 +- 其他状态（`PREPARE`、`UNKNOWN`、响应不可读）：持续轮询直到超过 `label_state_timeout_ms`，随后刷新失败，作业从上一个 checkpoint 重放该批数据。 + +该轮询在 checkpoint 路径上同步执行，且 sink 的重试循环最多会重复 `max_retries + 1` 次，最坏阻塞时间约为 `(max_retries + 1) * label_state_timeout_ms`。请将 `label_state_timeout_ms` 配置为小于作业的 checkpoint 超时时间，使未决事务尽早以 checkpoint 失败的形式暴露，而不是长时间阻塞 barrier。 + +由于失败的刷新会从 checkpoint 重放，同一窗口内先前已可见的批次可能被再次提交。Sink 保持至少一次语义：DUPLICATE KEY 表在重放后可能出现重复行，Primary Key 表可以吸收这些重复。 + ### Zeta 定时刷新 该引擎级能力仅由 Zeta 支持。可以在 `env` 块中配置 `sink.flush.interval`，使尚未达到 `batch_max_rows` 和 `batch_max_bytes` 的缓冲数据也能定时通过 StarRocks Stream Load 写出。Spark 和 Flink 不会触发该定时刷新。 diff --git a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitor.java b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitor.java index 1458a5643c..45b12a2d32 100644 --- a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitor.java +++ b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitor.java @@ -52,8 +52,6 @@ public class StarRocksStreamLoadVisitor { private long pos; private static final String RESULT_FAILED = "Fail"; private static final String RESULT_SUCCESS = "Success"; - private static final String RESULT_PUBLISH_TIMEOUT = "Publish Timeout"; - private static final String RESULT_LABEL_EXISTED = "Label Already Exists"; private static final String LABEL_STATE_VISIBLE = "VISIBLE"; private static final String LABEL_STATE_COMMITTED = "COMMITTED"; private static final String RESULT_LABEL_PREPARE = "PREPARE"; @@ -69,7 +67,7 @@ public class StarRocksStreamLoadVisitor { private final long labelStateTimeoutMs; public StarRocksStreamLoadVisitor(SinkConfig sinkConfig, TableSchema tableSchema) { - this(sinkConfig, tableSchema, new HttpHelper(sinkConfig), DEFAULT_LABEL_STATE_TIMEOUT_MS); + this(sinkConfig, tableSchema, new HttpHelper(sinkConfig), 0L); } /** @@ -78,12 +76,16 @@ public class StarRocksStreamLoadVisitor { */ StarRocksStreamLoadVisitor( SinkConfig sinkConfig, TableSchema tableSchema, HttpHelper httpHelper) { - this(sinkConfig, tableSchema, httpHelper, DEFAULT_LABEL_STATE_TIMEOUT_MS); + this(sinkConfig, tableSchema, httpHelper, 0L); } /** * Creates a visitor with explicit HTTP transport and label-state timeout for deterministic * boundary tests. + * + * <p>A non-positive {@code labelStateTimeoutMs} falls back to {@code + * sink_config.label_state_timeout_ms} and finally to the built-in default, so production + * behaviour stays configurable while tests keep an explicit override. */ StarRocksStreamLoadVisitor( SinkConfig sinkConfig, @@ -93,7 +95,14 @@ public class StarRocksStreamLoadVisitor { this.sinkConfig = sinkConfig; this.tableSchema = tableSchema; this.httpHelper = httpHelper; - this.labelStateTimeoutMs = Math.max(1, labelStateTimeoutMs); + long configuredTimeoutMs = labelStateTimeoutMs; + if (configuredTimeoutMs <= 0) { + configuredTimeoutMs = + sinkConfig.getLabelStateTimeoutMs() > 0 + ? sinkConfig.getLabelStateTimeoutMs() + : DEFAULT_LABEL_STATE_TIMEOUT_MS; + } + this.labelStateTimeoutMs = Math.max(1, configuredTimeoutMs); checkBatchMaxBytes(sinkConfig.getBatchMaxBytes(), sinkConfig.getBatchMaxSize()); } @@ -136,21 +145,11 @@ public class StarRocksStreamLoadVisitor { LOG.debug("StreamLoad response:\n" + JsonUtils.toJsonString(loadResult)); } Object resultStatus = loadResult.get(keyStatus); - if (RESULT_SUCCESS.equals(resultStatus) || RESULT_PUBLISH_TIMEOUT.equals(resultStatus)) { - return true; - } - if (RESULT_LABEL_EXISTED.equals(resultStatus)) { - LOG.debug("StreamLoad response:\n" + JsonUtils.toJsonString(loadResult)); - // The original request may already be committed, so never resend it under a new label - // until StarRocks reports the final state of the existing label. - checkLabelState(host, flushData.getLabel()); + if (RESULT_SUCCESS.equals(resultStatus)) { return true; } - if (RESULT_FAILED.equals(resultStatus)) { - if (isLabelAlreadyUsed(loadResult, flushData.getLabel())) { - checkLabelState(host, flushData.getLabel()); - return true; - } + if (RESULT_FAILED.equals(resultStatus) + && !isLabelAlreadyUsed(loadResult, flushData.getLabel())) { StringBuilder errorBuilder = new StringBuilder("Failed to flush data to StarRocks \n"); errorBuilder .append(sinkConfig.getDatabase()) @@ -177,10 +176,21 @@ public class StarRocksStreamLoadVisitor { throw new StarRocksConnectorException( StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, errorBuilder.toString()); } - throw new StarRocksConnectorException( - StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, - "Unable to flush data to StarRocks: unexpected result status. " - + JsonUtils.toJsonString(loadResult)); + // Publish Timeout, Label Already Exists, a Fail response blaming a reused label, or any + // other non-terminal status: StarRocks received the request but the transaction outcome is + // NOT confirmed. A publish that timed out can still be aborted (for example when the + // cluster was just restarted), so the batch may only be released after the label state is + // explicitly confirmed VISIBLE/COMMITTED; an ABORTED label authorizes a resend under a new + // label, and an unresolved label fails closed so the job replays from its checkpoint. + LOG.warn( + "StreamLoad for label[{}] on table[{}.{}] returned non-final status[{}];" + + " resolving the label state before releasing the batch.", + flushData.getLabel(), + sinkConfig.getDatabase(), + sinkConfig.getTable(), + resultStatus); + checkLabelState(host, flushData.getLabel(), loadResult); + return true; } private String getAvailableHost() { @@ -233,8 +243,24 @@ public class StarRocksStreamLoadVisitor { "Failed to join rows data, unsupported `format` from stream load properties:"); } + /** + * Resolves a non-final stream load outcome (Publish Timeout, Label Already Exists, reused-label + * failure, unrecognized status) to a terminal label state within one total deadline. + * + * <p>Only {@code VISIBLE}/{@code COMMITTED} releases the batch, only an explicit {@code + * ABORTED} authorizes the caller to retry under a new label, and everything else — {@code + * PREPARE}, {@code UNKNOWN}, or an unreadable response — keeps polling until the deadline and + * then fails closed. {@code UNKNOWN} is polled rather than failed immediately because a + * front-end that is still recovering from a restart may report {@code UNKNOWN} for a label + * whose transaction is in fact committed. + * + * @param host StarRocks front-end host that answered the stream load + * @param label label used by the current batch + * @param loadResult stream load response whose non-final outcome is being resolved + */ @SuppressWarnings("unchecked") - private void checkLabelState(String host, String label) throws IOException { + private void checkLabelState(String host, String label, Map<String, Object> loadResult) + throws IOException { int idx = 0; long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(labelStateTimeoutMs); while (System.nanoTime() < deadlineNanos) { @@ -251,24 +277,14 @@ public class StarRocksStreamLoadVisitor { queryLoadStateUrl, getLoadStateHttpHeader(label), remainingTimeoutMs(deadlineNanos)); - if (result == null) { - throw new StarRocksConnectorException( - StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, - String.format( - "Failed to flush data to StarRocks, Error " - + "could not get the final state of label[%s].\n", - label), - null); - } - String labelState = (String) result.get("state"); + String labelState = result == null ? null : (String) result.get("state"); if (null == labelState) { - throw new StarRocksConnectorException( - StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, - String.format( - "Failed to flush data to StarRocks, Error " - + "could not get the final state of label[%s]. response[%s]\n", - label, JsonUtils.toJsonString(result)), - null); + LOG.warn( + "Could not read the state of label[{}], response[{}]; will retry before deadline.", + label, + result); + sleepBeforeNextLabelCheck(++idx, deadlineNanos, label); + continue; } LOG.info(String.format("Checking label[%s] state[%s]\n", label, labelState)); switch (labelState) { @@ -288,23 +304,31 @@ public class StarRocksStreamLoadVisitor { true); case RESULT_LABEL_UNKNOWN: default: - throw new StarRocksConnectorException( - StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, - String.format( - "Failed to flush data to StarRocks, Error " - + "label[%s] state[%s]\n", - label, labelState)); + LOG.warn( + "Label[{}] state[{}] is not terminal; will retry before deadline.", + label, + labelState); + sleepBeforeNextLabelCheck(++idx, deadlineNanos, label); + continue; } } catch (IOException e) { - throw new StarRocksConnectorException( - StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, e); + LOG.warn( + "Failed to check StarRocks label [{}] state, will retry before deadline.", + label, + e); + sleepBeforeNextLabelCheck(++idx, deadlineNanos, label); } } throw new StarRocksConnectorException( StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, String.format( - "Timed out after %d ms while checking the final state of label[%s].", - labelStateTimeoutMs, label)); + "Timed out after %d ms while checking the final state of label[%s]" + + " returned for non-final stream load response Status[%s]" + + " Message[%s].", + labelStateTimeoutMs, + label, + loadResult.get("Status"), + loadResult.get("Message"))); } /** diff --git a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/SinkConfig.java b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/SinkConfig.java index 4b0ee5e6eb..49317512b9 100644 --- a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/SinkConfig.java +++ b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/SinkConfig.java @@ -65,6 +65,8 @@ public class SinkConfig implements Serializable { private int httpSocketTimeout; + private long labelStateTimeoutMs; + @Getter private final Map<String, Object> streamLoadProps = new HashMap<>(); public static SinkConfig of(ReadonlyConfig config) { @@ -98,6 +100,7 @@ public class SinkConfig implements Serializable { sinkConfig.setDataSaveMode(config.get(StarRocksSinkOptions.DATA_SAVE_MODE)); sinkConfig.setCustomSql(config.get(StarRocksSinkOptions.CUSTOM_SQL)); sinkConfig.setHttpSocketTimeout(config.get(StarRocksSinkOptions.HTTP_SOCKET_TIMEOUT_MS)); + sinkConfig.setLabelStateTimeoutMs(config.get(StarRocksSinkOptions.LABEL_STATE_TIMEOUT_MS)); return sinkConfig; } } diff --git a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/StarRocksSinkOptions.java b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/StarRocksSinkOptions.java index 5c20abd444..c3cdaacb2c 100644 --- a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/StarRocksSinkOptions.java +++ b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/config/StarRocksSinkOptions.java @@ -150,6 +150,16 @@ public class StarRocksSinkOptions extends StarRocksBaseOptions { .defaultValue(3 * 60 * 1000) .withDescription("Set http socket timeout, default is 3 minutes."); + public static final Option<Long> LABEL_STATE_TIMEOUT_MS = + Options.key("label_state_timeout_ms") + .longType() + .defaultValue(3 * 60 * 1000L) + .withDescription( + "Total time the sink polls get_load_state to resolve a non-final stream " + + "load outcome (Publish Timeout, Label Already Exists, reused-label " + + "failure) to VISIBLE/COMMITTED/ABORTED before failing the flush " + + "and letting the job replay from its checkpoint, default is 3 minutes."); + public static final Option<String> CUSTOM_SQL = Options.key("custom_sql") .stringType() diff --git a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkFactory.java b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkFactory.java index a2b0c4e06d..9ea4726149 100644 --- a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkFactory.java +++ b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkFactory.java @@ -71,6 +71,9 @@ public class StarRocksSinkFactory implements TableSinkFactory { MULTI_TABLE_SINK_REPLICA, StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE, StarRocksSinkOptions.HTTP_SOCKET_TIMEOUT_MS) + .optional( + StarRocksSinkOptions.LABEL_STATE_TIMEOUT_MS, + Conditions.greaterThan(StarRocksSinkOptions.LABEL_STATE_TIMEOUT_MS, 0L)) .optional( SinkConnectorCommonOptions.TABLE_OPTIONS, Conditions.extension( diff --git a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitorTest.java b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitorTest.java index 9d81883773..dd3af24799 100644 --- a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitorTest.java +++ b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitorTest.java @@ -23,7 +23,9 @@ import org.apache.seatunnel.connectors.seatunnel.starrocks.exception.StarRocksCo import org.junit.jupiter.api.Test; +import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -32,6 +34,7 @@ import java.util.Map; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; @@ -39,6 +42,7 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -123,22 +127,82 @@ public class StarRocksStreamLoadVisitorTest { } /** - * Verifies that a successful load awaiting publication is not submitted a second time. + * Verifies that Publish Timeout is resolved through the label state instead of being trusted as + * success. * - * <p>StarRocks documents Publish Timeout as loaded successfully and requiring no retry. + * <p>StarRocks documents Publish Timeout as "loaded, publish pending", but a cluster that was + * just restarted can still abort the pending transaction. Releasing the batch without + * confirming the label state silently loses it, so the visitor must poll get_load_state and + * only a VISIBLE/COMMITTED label confirms the load. */ @Test - void returnsSuccessForPublishTimeout() throws Exception { + void resolvesPublishTimeoutByConfirmingLabelState() throws Exception { SinkConfig sinkConfig = createSinkConfig(); HttpHelper httpHelper = mock(HttpHelper.class); when(httpHelper.tryHttpConnection("http://localhost:8030";)).thenReturn(true); when(httpHelper.doHttpPut(anyString(), any(byte[].class), any())) .thenReturn(createLoadResult("Publish Timeout")); + Map<String, Object> labelState = new HashMap<>(); + labelState.put("state", "VISIBLE"); + when(httpHelper.doHttpGet(anyString(), any(), anyInt())).thenReturn(labelState); StarRocksStreamLoadVisitor visitor = new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper); assertTrue(visitor.doStreamLoad(createFlushTuple())); - verify(httpHelper, never()).doHttpGet(anyString(), any(), anyInt()); + verify(httpHelper).doHttpGet(anyString(), any(), anyInt()); + } + + /** + * Verifies that a Publish Timeout label reported ABORTED authorizes a replacement label. + * + * <p>An aborted transaction means the batch never became durable, so the manager must resend it + * under a new label instead of dropping it. + */ + @Test + void recreatesLabelWhenPublishTimeoutLabelIsAborted() throws Exception { + SinkConfig sinkConfig = createSinkConfig(); + HttpHelper httpHelper = mock(HttpHelper.class); + when(httpHelper.tryHttpConnection("http://localhost:8030";)).thenReturn(true); + when(httpHelper.doHttpPut(anyString(), any(byte[].class), any())) + .thenReturn(createLoadResult("Publish Timeout")); + Map<String, Object> labelState = new HashMap<>(); + labelState.put("state", "ABORTED"); + when(httpHelper.doHttpGet(anyString(), any(), anyInt())).thenReturn(labelState); + StarRocksStreamLoadVisitor visitor = + new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper); + + StarRocksConnectorException exception = + assertThrows( + StarRocksConnectorException.class, + () -> visitor.doStreamLoad(createFlushTuple())); + assertTrue(exception.needReCreateLabel()); + } + + /** + * Verifies that a Publish Timeout label stuck in PREPARE fails closed after the bounded wait. + * + * <p>An unresolved outcome must never release the batch: the thrown error lets the manager keep + * the buffered rows so the job can replay them from its checkpoint. + */ + @Test + void failsWhenPublishTimeoutLabelNeverResolves() throws Exception { + SinkConfig sinkConfig = createSinkConfig(); + HttpHelper httpHelper = mock(HttpHelper.class); + when(httpHelper.tryHttpConnection("http://localhost:8030";)).thenReturn(true); + when(httpHelper.doHttpPut(anyString(), any(byte[].class), any())) + .thenReturn(createLoadResult("Publish Timeout")); + Map<String, Object> labelState = new HashMap<>(); + labelState.put("state", "PREPARE"); + when(httpHelper.doHttpGet(anyString(), any(), anyInt())).thenReturn(labelState); + StarRocksStreamLoadVisitor visitor = + new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper, 500); + + StarRocksConnectorException exception = + assertThrows( + StarRocksConnectorException.class, + () -> visitor.doStreamLoad(createFlushTuple())); + assertFalse(exception.needReCreateLabel()); + verify(httpHelper, atLeastOnce()).doHttpGet(anyString(), any(), anyInt()); } /** @@ -257,7 +321,11 @@ public class StarRocksStreamLoadVisitorTest { } /** - * Verifies that an unknown label state fails closed without authorizing a replacement label. + * Verifies that an UNKNOWN label state is polled until the deadline before failing closed. + * + * <p>A front-end recovering from a restart can report UNKNOWN for a committed label, so an + * immediate failure would cause needless replay; the state must be retried inside the bounded + * wait and still never authorize a replacement label. */ @Test void failsWhenExistingLabelStateIsUnknown() throws Exception { @@ -270,30 +338,38 @@ public class StarRocksStreamLoadVisitorTest { labelState.put("state", "UNKNOWN"); when(httpHelper.doHttpGet(anyString(), any(), anyInt())).thenReturn(labelState); StarRocksStreamLoadVisitor visitor = - new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper); + new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper, 500); StarRocksConnectorException exception = assertThrows( StarRocksConnectorException.class, () -> visitor.doStreamLoad(createFlushTuple())); assertFalse(exception.needReCreateLabel()); + verify(httpHelper, atLeastOnce()).doHttpGet(anyString(), any(), anyInt()); } /** - * Verifies that an unrecognized response status cannot be converted into a successful flush. + * Verifies that an unrecognized response status is resolved through the label state. + * + * <p>Any 200 response with a Status field means StarRocks accepted the request, so its + * transaction exists and can be confirmed; trusting an unknown status string would either drop + * or duplicate the batch. */ @Test - void failsForUnexpectedStreamLoadStatus() throws Exception { + void resolvesUnexpectedStreamLoadStatusThroughLabelState() throws Exception { SinkConfig sinkConfig = createSinkConfig(); HttpHelper httpHelper = mock(HttpHelper.class); when(httpHelper.tryHttpConnection("http://localhost:8030";)).thenReturn(true); when(httpHelper.doHttpPut(anyString(), any(byte[].class), any())) .thenReturn(createLoadResult("Mystery")); + Map<String, Object> labelState = new HashMap<>(); + labelState.put("state", "VISIBLE"); + when(httpHelper.doHttpGet(anyString(), any(), anyInt())).thenReturn(labelState); StarRocksStreamLoadVisitor visitor = new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper); - assertThrows( - StarRocksConnectorException.class, () -> visitor.doStreamLoad(createFlushTuple())); + assertTrue(visitor.doStreamLoad(createFlushTuple())); + verify(httpHelper, atLeastOnce()).doHttpGet(anyString(), any(), anyInt()); } /** @@ -338,7 +414,7 @@ public class StarRocksStreamLoadVisitorTest { new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper, 2000); assertTrue(visitor.doStreamLoad(createFlushTuple())); - verify(httpHelper, org.mockito.Mockito.times(2)).doHttpGet(anyString(), any(), anyInt()); + verify(httpHelper, times(2)).doHttpGet(anyString(), any(), anyInt()); } /** @@ -361,7 +437,7 @@ public class StarRocksStreamLoadVisitorTest { new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper, 2000); assertTrue(visitor.doStreamLoad(createFlushTuple())); - verify(httpHelper, org.mockito.Mockito.times(2)).doHttpGet(anyString(), any(), anyInt()); + verify(httpHelper, times(2)).doHttpGet(anyString(), any(), anyInt()); } /** @@ -379,13 +455,68 @@ public class StarRocksStreamLoadVisitorTest { preparing.put("state", "PREPARE"); when(httpHelper.doHttpGet(anyString(), any(), anyInt())).thenReturn(preparing); StarRocksStreamLoadVisitor visitor = - new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper, 10); + new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper, 500); assertThrows( StarRocksConnectorException.class, () -> visitor.doStreamLoad(createFlushTuple())); verify(httpHelper, atLeastOnce()).doHttpGet(anyString(), any(), anyInt()); } + /** + * Verifies that the {@code label_state_timeout_ms} sink configuration drives the polling + * deadline when the visitor is built through the production constructor. + * + * <p>The bounded assertion keeps the test fast even if a regression falls back to the built-in + * default deadline. + */ + @Test + void usesConfiguredLabelStateTimeoutFromSinkConfig() throws Exception { + SinkConfig sinkConfig = createSinkConfig(); + when(sinkConfig.getLabelStateTimeoutMs()).thenReturn(300L); + HttpHelper httpHelper = mock(HttpHelper.class); + when(httpHelper.tryHttpConnection("http://localhost:8030";)).thenReturn(true); + when(httpHelper.doHttpPut(anyString(), any(byte[].class), any())) + .thenReturn(createLoadResult("Publish Timeout")); + Map<String, Object> labelState = new HashMap<>(); + labelState.put("state", "PREPARE"); + when(httpHelper.doHttpGet(anyString(), any(), anyInt())).thenReturn(labelState); + StarRocksStreamLoadVisitor visitor = + new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper); + + StarRocksConnectorException exception = + assertTimeoutPreemptively( + Duration.ofSeconds(10), + () -> + assertThrows( + StarRocksConnectorException.class, + () -> visitor.doStreamLoad(createFlushTuple()))); + assertFalse(exception.needReCreateLabel()); + verify(httpHelper, atLeastOnce()).doHttpGet(anyString(), any(), anyInt()); + } + + /** + * Verifies that a transient failure while reading the label state is retried inside the bounded + * wait instead of failing the flush immediately. + */ + @Test + void retriesLabelStateCheckWhenHttpGetFails() throws Exception { + SinkConfig sinkConfig = createSinkConfig(); + HttpHelper httpHelper = mock(HttpHelper.class); + when(httpHelper.tryHttpConnection("http://localhost:8030";)).thenReturn(true); + when(httpHelper.doHttpPut(anyString(), any(byte[].class), any())) + .thenReturn(createLoadResult("Publish Timeout")); + Map<String, Object> labelState = new HashMap<>(); + labelState.put("state", "VISIBLE"); + when(httpHelper.doHttpGet(anyString(), any(), anyInt())) + .thenThrow(new IOException("connection reset")) + .thenReturn(labelState); + StarRocksStreamLoadVisitor visitor = + new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper, 2000); + + assertTrue(visitor.doStreamLoad(createFlushTuple())); + verify(httpHelper, times(2)).doHttpGet(anyString(), any(), anyInt()); + } + /** * Creates the minimal deterministic sink configuration needed by response-handling tests. *
