实时同步kafka消息至Doris,出现DorisConnectorException: [stream load error] - HTTP/1.1 307 Temporary Redirect异常
- Dominant language
- Java
- Stars
- 9.7k
- Forks
- 2.4k
- Avg merge
- 3d 17h
- Merged PRs (30d)
- 210
Description
### Search before asking
- [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues.
### What happened
使用kafka实时同步数据至doris,doris有2个fe 3个be,出现DorisConnectorException: [stream load error] - HTTP/1.1 307 Temporary Redirect异常,观察监控doris机器在该时间段正常、无相关错误日志
在https://github.com/apache/seatunnel/issues/9355内该提问者反馈说可在sink fenodes配置多个fe地址解决该问题,不过我已将两个fe均加入配置内
### SeaTunnel Version
2.3.8
### SeaTunnel Config
```conf
env {
"job.mode"=STREAMING
"job.name"="SeaTunnel_Job"
"savemode.execute.location"=CLUSTER
}
source {
Kafka {
"start_mode"="early"
"commit_on_checkpoint"="false"
format=JSON
"debezium_record_include_schema"="true"
parallelism="1"
"start_mode.timestamp"=null
"consumer.group"="groupName"
"partition-discovery.interval-millis"="-1"
"result_table_name"=Table18962778563232
topic="topicName"
schema {
fields {
thingid=STRING
timestamp=STRING
"iot_key"=STRING
propertydata=STRING
}
table="default.iot_property_data7"
}
"bootstrap.servers"="xx.xx.xx.xx:9092"
}
}
sink {
Doris {
tableName="table_name"
"data_save_mode"="APPEND_DATA"
"schema_save_mode"="CREATE_SCHEMA_WHEN_NOT_EXIST"
"multi_table_sink_replica"="1"
"save_mode_create_template"="CREATE TABLE IF NOT EXISTS `${database}`.`${table}` (${rowtype_fields}) ENGINE=OLAP PROPERTIES (\"replication_allocation\" = \"tag.location.default: 1\",\"in_memory\" = \"false\",\"storage_format\" = \"V2\",\"disable_auto_compaction\" = \"false\")"
"needs_unsupported_type_casting"="false"
"sink.label-prefix"=c8d
"doris.config" {
"read_json_by_line"=true
format=json
}
"doris.batch.size"="1024"
"sink.enable-2pc"="false"
"sink.enable-delete"="false"
"source_table_name"=Table18962778563233
database="database_name"
table="table_name"
password="xxxx"
username=xxxx
fenodes="1.2.3.4:8030,1.2.3.5:8030"
query-port="9030"
}
}
```
### Running Command
```shell
bin/seatunnel.sh --config doris.json
```
### Error Exception
```log
java.lang.RuntimeException: java.lang.RuntimeException: org.apache.seatunnel.connectors.doris.exception.DorisConnectorException: ErrorCode:[Doris-01], ErrorDescription:[stream load error]
at org.apache.seatunnel.engine.server.task.flow.SinkFlowLifeCycle.received(SinkFlowLifeCycle.java:253)
at org.apache.seatunnel.engine.server.task.flow.SinkFlowLifeCycle.received(SinkFlowLifeCycle.java:66)
at org.apache.seatunnel.engine.server.task.SeaTunnelTransformCollector.collect(SeaTunnelTransformCollector.java:39)
at org.apache.seatunnel.engine.server.task.SeaTunnelTransformCollector.collect(SeaTunnelTransformCollector.java:27)
at org.apache.seatunnel.engine.server.task.group.queue.IntermediateBlockingQueue.handleRecord(IntermediateBlockingQueue.java:75)
at org.apache.seatunnel.engine.server.task.group.queue.IntermediateBlockingQueue.collect(IntermediateBlockingQueue.java:50)
at org.apache.seatunnel.engine.server.task.flow.IntermediateQueueFlowLifeCycle.collect(IntermediateQueueFlowLifeCycle.java:51)
at org.apache.seatunnel.engine.server.task.TransformSeaTunnelTask.collect(TransformSeaTunnelTask.java:73)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.stateProcess(SeaTunnelTask.java:168)
at org.apache.seatunnel.engine.server.task.TransformSeaTunnelTask.call(TransformSeaTunnelTask.java:78)
at org.apache.seatunnel.engine.server.TaskExecutionService$BlockingWorker.run(TaskExecutionService.java:684)
at org.apache.seatunnel.engine.server.TaskExecutionService$NamedTaskWrapper.run(TaskExecutionService.java:1009)
at org.apache.seatunnel.api.tracing.MDCRunnable.run(MDCRunnable.java:39)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: org.apache.seatunnel.connectors.doris.exception.DorisConnectorException: ErrorCode:[Doris-01], ErrorDescription:[stream load error]
at org.apache.seatunnel.api.sink.multitablesink.MultiTableSinkWriter.subSinkErrorCheck(MultiTableSinkWriter.java:136)
at org.apache.seatunnel.api.sink.multitablesink.MultiTableSinkWriter.write(MultiTableSinkWriter.java:173)
at org.apache.seatunnel.api.sink.multitablesink.MultiTableSinkWriter.write(MultiTableSinkWriter.java:47)
at org.apache.seatunnel.engine.server.task.flow.SinkFlowLifeCycle.received(SinkFlowLifeCycle.java:249)
... 17 more
Caused by: org.apache.seatunnel.connectors.doris.exception.DorisConnectorException: ErrorCode:[Doris-01], ErrorDescription:[stream load error]
at org.apache.seatunnel.connectors.doris.sink.writer.DorisStreamLoad.stopLoad(DorisStreamLoad.java:228)
at org.apache.seatunnel.connectors.doris.sink.writer.DorisSinkWriter.flush(DorisSinkWriter.java:156)
at org.apache.seatunnel.connectors.doris.sink.writer.DorisSinkWriter.write(DorisSinkWriter.java:136)
at org.apache.seatunnel.connectors.doris.sink.writer.DorisSinkWriter.write(DorisSinkWriter.java:53)
at org.apache.seatunnel.api.sink.multitablesink.MultiTableWriterRunnable.run(MultiTableWriterRunnable.java:62)
... 6 more
Caused by: org.apache.seatunnel.connectors.doris.exception.DorisConnectorException: ErrorCode:[Doris-01], ErrorDescription:[stream load error] - HTTP/1.1 307 Temporary Redirect
at org.apache.seatunnel.connectors.doris.sink.writer.DorisStreamLoad.handlePreCommitResponse(DorisStreamLoad.java:217)
at org.apache.seatunnel.connectors.doris.sink.writer.DorisStreamLoad.stopLoad(DorisStreamLoad.java:226)
... 10 more
```
### Zeta or Flink or Spark Version
_No response_
### Java or Scala Version
_No response_
### Screenshots
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with connectors/doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisStreamLoad.java, especially handlePreCommitResponse and stopLoad, then trace the call from DorisSinkWriter.flush. Reproduce the Kafka-to-Doris streaming configuration with multiple fenodes and inspect the redirect response; done means the failure is reproduced and its handling or configuration cause is documented or corrected.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- data-engineering, databases
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 35/100