[Bug]: StarRocks Writer中postSql 提前执行导致 StreamLoad 尚未完成，目标表数据缺失

[Sun0227]

1️⃣ 问题描述 | Problem Description

问题描述：
当使用 starrockswriter 将数据写入 StarRocks 的临时表后，在 writer 的 postSql 中执行 INSERT INTO target SELECT * FROM tmp，发现 target 表中数据不完整，部分未导入。

初步排查是因为 StarRocks Stream Load 是异步提交，虽然 writer.flush() 被调用，但实际导入任务未完成时 postSql 已开始执行。

复现方式：

设置 writer 将数据导入 StarRocks 的中间表（如 tmp）

配置 postSql 执行 INSERT INTO real SELECT * FROM tmp

发现 real 表中行数小于 tmp 表

期望行为：
Addax 应确保所有 stream load 完成（即 SHOW LOAD WHERE label = ... 状态为 FINISHED）后再执行 postSql

建议修复方式：

支持在 writer 插件中开启 waitForStreamLoadComplete 开关；

或支持用户传入 labelPrefix，并在 close 时显式轮询 SHOW LOAD 状态；

或允许延迟执行 postSql，外部控制

starrockswriter 版本：4.1.4

2️⃣ 当前使用的 Addax 版本 | Addax Version

4.2.0 or older

3️⃣ 操作系统类型 | Operating System

None

4️⃣ 使用的 Java JDK 版本 | Java JDK Version

Oracle JDK 1.8.0

5️⃣ 相关日志/输出结果 | Relevant Log Output

[wgzhao]

1. At what data volume does this abnormal situation you mentioned occur? In my testing, with approximately 100,000 records, I haven't encountered this issue.

2. Does this anomaly still occur in the latest released version?

[Sun0227]

1.场景如下：单表1000w数据，10个表同时往同一个StarRocks中抽取数据。
该场景下几乎必现，原因可能如下：
StarRocks的StreamLoad模式，频繁导入数据导致StarRocks服务端负载较高，数据可见性异步。StreamLoad提交成功但是目的表里数据不完整，此时执行postSql从该表里查数据就会导致该问题产生。
所以，在执行postSql前，可以轮询StreamLoad的Label，等该任务真正完成后，再执行postSql可以解决该问题
2.用6.0.2仍然有该问题

[wgzhao]

Yes, when asynchronously calling the stream load interface, it determines whether to proceed based on the returned status. In the streamLoad method of the DorisStreamLoadObserver class, there is the following code:

final String keyStatus = "Status";
if (null == loadResult || !loadResult.containsKey(keyStatus)) {
    throw new IOException("Unable to flush data to Doris: unknown result status.");
}
LOG.debug("StreamLoad response:{}", JSON.toJSONString(loadResult));
if (RESULT_FAILED.equals(loadResult.get(keyStatus))) {
    throw new IOException(
            "Failed to flush data to Doris.\n" + JSON.toJSONString(loadResult)
    );
}
else if (RESULT_LABEL_EXISTED.equals(loadResult.get(keyStatus))) {
    LOG.debug("StreamLoad response:{}", JSON.toJSONString(loadResult));
    checkStreamLoadState(host, data.getLabel());
}

According to the official documentation regarding the Status explanation, only when the value is Label Already Exists does it indicate that the data import might not be completed yet, requiring a subsequent status check. Therefore, the checkStreamLoadState method is called to verify whether the data import for the current label has been completed.

[Sun0227]

上述代码是在startAsyncFlushing()线程中执行的。postSql是在另一个线程中执行，并不会阻塞等待每一个Label执行完成。就会造成post已经执行了，上述代码还没有执行。
下面case是我在postSql中直接删除目标表中的报错：
[Thread-232] DEBUG c.w.a.p.w.s.manager.StarRocksStreamLoadVisitor:73 - StreamLoad response: {"TxnId":-1,"Label":"18dbc585-4a37-4b7a-92d9-52dc78db89b3","Status":"Fail","Message":"unknown table \"pms_test.temp_dp_e_794993\"","NumberTotalRows":0,"NumberLoadedRows":0,"NumberFilteredRows":0,"NumberUnselectedRows":0,"LoadBytes":0,"LoadTimeMs":0,"BeginTxnTimeMs":0,"StreamLoadPlanTimeMs":0,"ReadDataTimeMs":0,"WriteDataTimeMs":0,"CommitAndPublishTimeMs":0}

[wgzhao]

@Sun0227 Could you verify if this issue has been fixed with the updated code?

[Sun0227]

不好意思，我们运行环境目前不支持java17，不太容易用新分支去验证。
我看代码新增了在Task#post里checkLabelState，但是postSql是在Job#post里执行的，同时Task#post和Job#post分别在2个线程里执行的，如果框架保证Job#post一定在Task#pos后执行，应该就没问题了，否则还是有该问题存在

[wgzhao]

Yes, the Task#post method must be called before Job#post.