apache / apache/seatunnel

实时同步kafka消息至Doris,出现DorisConnectorException: [stream load error] - HTTP/1.1 307 Temporary Redirect异常

Open
#9,871 4 comments 0 reactions 0 assignees View on GitHub
bug
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.