flink sink doris不能成功

[haiziguo]

be已经设置 disable_stream_load_2pc=false
使用
https://github.com/apache/doris/blob/master/samples/doris-demo/flink-demo-v1.1/src/main/java/org/apache/doris/demo/flink/DorisFlinkConnectorDemoV1.java

报400错误

[haiziguo]

具体的描述如下
org.apache.doris.flink.sink.DorisSink#createWriter ->org.apache.doris.flink.sink.writer.DorisWriter#initializeLoad->org.apache.doris.flink.sink.writer.DorisStreamLoad#abortPreCommit

走到abortTransaction 报http 400错误 程序终止

[JNSimba]

What is your upstream? Is checkpoint enabled? Can you provide the relevant taskmanager logs?

[haiziguo]

use the sample demo V1
code :

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
env.enableCheckpointing(10000);
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, Time.milliseconds(30000)));

    DorisSink.Builder<String> builder = DorisSink.builder();
    final DorisReadOptions.Builder readOptionBuilder = DorisReadOptions.builder();
    Properties pro = new Properties();
    pro.setProperty("format", "json");
    pro.setProperty("read_json_by_line", "true");
    pro.setProperty("line_delimiter", "\n");
    DorisOptions.Builder dorisBuilder = DorisOptions.builder();
    dorisBuilder.setFenodes("10.10.20.83:8030")
            .setTableIdentifier("db_test.user_test4")
            .setUsername("xxx")
            .setPassword("xxxx");
    DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
    executionBuilder
            .setStreamLoadProp(pro)
            //.disable2PC()
            .setLabelPrefix("abc500");


    builder.setDorisReadOptions(readOptionBuilder.build())
            .setDorisExecutionOptions(executionBuilder.build())
            .setSerializer(new SimpleStringSerializer())

            .setDorisOptions(dorisBuilder.build());
    **List<String> list= Lists.newArrayList();
    for (int i = 100; i < 10000; i++) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("user_id",i);
        jsonObject.put("user_name","abc"+i);
        list.add(jsonObject.toJSONString());
    }
    String join = String.join("\n", list);**
   // System.out.println(join);
    env.fromElements(join).sinkTo(builder.build());

    env.execute("flink demo");

[haiziguo]

Send request to Doris FE 'http://10.10.20.83:8030/api/backends?is_alive=true' with user 'root'.
15:31:08,277 INFO org.apache.doris.flink.sink.writer.DorisWriter [367] - Backend Info:{"backends":[{"ip":"10.10.20.83","http_port":8040,"is_alive":true}]}
15:31:08,321 INFO org.apache.doris.flink.sink.writer.RecordBuffer [43] - init RecordBuffer capacity 1048576, count 3
15:31:08,323 INFO org.apache.doris.flink.sink.writer.DorisStreamLoad [135] - abort for labelSuffix abc500_0. start chkId 1.
15:31:08,689 INFO org.apache.doris.flink.sink.writer.DorisStreamLoad [204] - load Result {
"TxnId": 1607050,
"Label": "abc500_0_1",
"TwoPhaseCommit": "true",
"Status": "Success",
"Message": "OK",
"NumberTotalRows": 0,
"NumberLoadedRows": 0,
"NumberFilteredRows": 0,
"NumberUnselectedRows": 0,
"LoadBytes": 0,
"LoadTimeMs": 17,
"BeginTxnTimeMs": 1,
"StreamLoadPutTimeMs": 2,
"ReadDataTimeMs": 0,
"WriteDataTimeMs": 3,
"CommitAndPublishTimeMs": 0
}

15:31:11,717 INFO org.apache.doris.flink.sink.writer.DorisStreamLoad [168] - abort 1607050 for check label abc500_0_1.
15:31:12,861 WARN org.apache.doris.flink.sink.writer.DorisStreamLoad [265] - abort transaction response: HTTP/1.1 400 Bad Request
15:31:12,862 WARN org.apache.doris.flink.sink.writer.DorisStreamLoad [174] - failed to stream load data
org.apache.doris.flink.exception.DorisRuntimeException: Fail to abort transaction 1607050 with url http://10.10.20.83:8040/api/db_test/_stream_load_2pc
at org.apache.doris.flink.sink.writer.DorisStreamLoad.abortTransaction(DorisStreamLoad.java:266)
at org.apache.doris.flink.sink.writer.DorisStreamLoad.abortPreCommit(DorisStreamLoad.java:169)
at org.apache.doris.flink.sink.writer.DorisWriter.initializeLoad(DorisWriter.java:111)
at org.apache.doris.flink.sink.DorisSink.createWriter(DorisSink.java:64)
at org.apache.flink.streaming.runtime.operators.sink.SinkOperator.initializeState(SinkOperator.java:148)
at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:119)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:286)
at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:109)
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:711)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:687)
at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:654)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:927)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)

[JNSimba]

which version of doris is it?

[haiziguo]

doris V1.1.0
flink 1.14

[JNSimba]

You can test this api http://10.10.20.83:8040/api/db_test/_stream_load_2pc It can be accessed normally, refer here https://doris.apache.org/zh-CN/docs/data-operate/import/import-way/stream-load-manual/?_highlight=two_phase_commit