创建的若干kafka导入任务，部分成功导入，部分不成功

[warlockhades]

Doris version 0.12.0-rc03

成功指生产者send的消息写入了doris表中，不成功指生产者send的消息没有写入doris表中

某个不成功的任务创建前后都创建过成功的任务，应该不是服务器配置改动造成的

最近创建的任务都不成功

所有任务使用的都是同一个kafka集群
我在日志中看到提示kafka consume timeout，但在doris所在服务器可以消费到kafka消息，附件是日志

建表语句
create table test_doris_two (
cola INT NOT NULL,
colb VARCHAR(30),
colc CHAR(3)
)
UNIQUE KEY(cola)
DISTRIBUTED BY HASH(cola) BUCKETS 16;

创建topic
./bin/kafka-topics.sh --create --zookeeper 192.168.0.131:2181,192.168.0.132:2181,192.168.0.133:2181 --replication-factor 3 --partitions 3 --topic test_doris_two

创建任务
CREATE ROUTINE LOAD my_db.job_test_doris_two ON test_doris_two
COLUMNS TERMINATED BY "|",
COLUMNS(cola, colb, colc)
PROPERTIES
(
"desired_concurrent_number"="3",
"max_batch_interval" = "20",
"max_batch_rows" = "300000",
"max_batch_size" = "209715200",
"max_error_number" = "30000",
"strict_mode" = "false"
)
FROM KAFKA
(
"kafka_broker_list" = "192.168.0.131:9092,192.168.0.132:9092,192.168.0.133:9092",
"kafka_topic" = "test_doris_two",
"property.group.id" = "test_doris_two_consumers"
);

发送的某条数据
123|hahaha|a

没有写入数据库

doris.log

[warlockhades]

show routine load \G;
*************************** 7. row ***************************
Id: 1225395
Name: job_test_doris_two
CreateTime: 2021-10-13 19:47:19
PauseTime: N/A
EndTime: N/A
DbName: default_cluster:athena
TableName: test_doris_two
State: RUNNING
DataSourceType: KAFKA
CurrentTaskNum: 3
JobProperties: {"partitions":"","columnToColumnExpr":"cola,colb,colc","maxBatchIntervalS":"20","whereExpr":"","maxBatchSizeBytes":"209715200","columnSeparator":"'|'","maxErrorNum":"30000","currentTaskConcurrentNum":"3","maxBatchRows":"300000"}
DataSourceProperties: {"topic":"test_doris_two","currentKafkaPartitions":"0,1,2","brokerList":"192.168.0.131:9092,192.168.0.132:9092,192.168.0.133:9092"}
CustomProperties: {"group.id":"test_doris_two_consumers"}
Statistic: {"receivedBytes":0,"errorRows":0,"committedTaskNum":0,"loadedRows":0,"loadRowsRate":0,"abortedTaskNum":864,"totalRows":0,"unselectedRows":0,"receivedBytesRate":0,"taskExecuteTimeMs":1}
Progress: {"0":"OFFSET_END","1":"OFFSET_END","2":"OFFSET_END"}
ReasonOfStateChanged:
ErrorLogUrls:
OtherMsg:

[EmmyMiao87]

you could change your progress to offset_begin

[warlockhades]

你是说加一行"property.kafka_default_offset" = "0"？
但我需要消费任务创建后发送的消息

CREATE ROUTINE LOAD my_db.job_test_doris_two ON test_doris_two
COLUMNS TERMINATED BY "|",
COLUMNS(cola, colb, colc)
PROPERTIES
(
"desired_concurrent_number"="3",
"max_batch_interval" = "20",
"max_batch_rows" = "300000",
"max_batch_size" = "209715200",
"max_error_number" = "30000",
"strict_mode" = "false"
)
FROM KAFKA
(
"kafka_broker_list" = "192.168.0.131:9092,192.168.0.132:9092,192.168.0.133:9092",
"kafka_topic" = "test_doris_two",
"property.group.id" = "test_doris_two_consumers",
"property.kafka_default_offset" = "0"
);