【routine load】kafka 消息有堆积的情况下为什么仍然存在 abortedTask

[LangYuanzh]

环境信息 ：

1. kafka topic 3个patition
2. doris 1fe 3be
3. routine load：PROPERTIES (
   "desired_concurrent_number" = "3"
   )
   问题描述：
   通过 kafka 导入 100w条数据，kafka消息一直有堆积的情况下，仍然存在abortedTask
   任务初始状态：
   "committedTaskNum":16,"loadedRows":160759,"loadRowsRate":1000,"abortedTaskNum":8,"totalRows":160759,
   任务中间状态：
   "committedTaskNum":48,"loadedRows":578200,"loadRowsRate":2000,"abortedTaskNum":14,"totalRows":578200,
   期间kafka消息一直存在堆积，但是abortedTaskNum仍然增加到了14。

[morningman]

Which Doris version?
And could you provide related logs in be.INFO, like kafka consume timeout?

[LangYuanzh]

I re-tested and the situation reappeared.
The version of DORIS is: 0.14.0
The test procedure is as follows.

1. write 200w of data to kafka

2. Create a new routine load

3. abortedTaskNum goes up to 17, while there is still at least 20w of lag in each partition of the kafka topic
   {"receivedBytes":3854689109, "errorRows":0, "committedTaskNum":43, "loadedRows":478792, "loadRowsRate":1000, "abortedTaskNum":17," totalRows":478792,

4. Check the INFO log and two "kafka consume timeout" items were found during this time：
   I1123 09:52:42.717406 4126 data_consumer.cpp:214] kafka consume timeout: ec41fce5a0be25c3-01e964c3a8a00580
   I1123 09:53:44.608278 4262 data_consumer.cpp:214] kafka consume timeout: 3d4f61692cb092aa-9e6fbbe75c095bb9

[morningman]

This is strange. kafka consume timeout always means the kafka client can not fetch any data from kafka broker.
And this is directly return by librdkafka, which is the official c++ client of kafka.

You can try to set a unique group.id in routine load job to see if it can be solved.

[LangYuanzh]

In the test the group.id is set to 'test_consumer', the routine job is create as :
CREATE ROUTINE LOAD test.kafka_load_job ON test_table
with APPEND
COLUMNS ('all colums of the table')
PROPERTIES (
"desired_concurrent_number" = "3",
"max_batch_interval" = "5",
"max_batch_rows" = "200000",
"max_batch_size" = "209715200",
"format" = "json",
"timezone" = "Europe/London",
"json_root" = "$.payload.after"
)
FROM KAFKA(
"kafka_broker_list" = "my_kafka_broker",
"kafka_topic" = "test_topic",
"kafka_partitions" = "0,1,2",
"kafka_offsets" = "999999,999999,999999",
"property.group.id" = "test_sonsumer",
"property.kafka_default_offsets" = "OFFSET_END" );
There is another question, in the INFO log only two "kafka consume timeout" items were found but there was 17 abortedTask. Is there any other reason can casue a task be aborted?