[Bug] [Doris Sink] In Zeta mode, MySQL CDC cannot automatically create tables in Doris. The Doris version is 3.0+ and there is an error.
- Dominant language
- Java
- Stars
- 9.7k
- Forks
- 2.4k
- Avg merge
- 3d 13h
- Merged PRs (30d)
- 203
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
Zeta模式下MySQL CDC到Doris无法自动建表,Doris版本3.0+,且报错
### SeaTunnel Version
2.3.10
### SeaTunnel Config
```conf
env {
parallelism = 1
job.mode = "STREAMING"
checkpoint.interval = 10000
}
source {
MySQL-CDC {
base-url = "jdbc:mysql://192.168.121.26:3306/data_api"
username = "root"
database-names=["data_api"]
password = "1234@qwer"
table-pattern="data_api.*"
schema-changes.enabled="true"
server-id = "10086-10186"
startup.mode="INITIAL"
}
}
sink {
Doris {
fenodes = "---"
username = admin
password = "*****"
database = "ms_sync"
table="z_${table_name}"
schema_save_mode="CREATE_SCHEMA_WHEN_NOT_EXIST"
data_save_mode="APPEND_DATA"
sink.enable-delete = "true"
sink.label-prefix = "test1"
sink.max-retries=3
doris.config {
format = "json"
read_json_by_line = "true"
}
}
}
```
### Running Command
```shell
-
```
### Error Exception
```log
worker2 | java.io.IOException: org.apache.seatunnel.connectors.doris.exception.DorisConnectorException: ErrorCode:[Doris-01], ErrorDescription:[stream load error] - [NOT_FOUND]TStatus: errCode = 7, detailMessage = table not found, tableName=z_auth_role
worker2 | at org.apache.seatunnel.connectors.doris.sink.writer.RecordBuffer.stopBufferData(RecordBuffer.java:96) ~[?:?]
worker2 | at org.apache.seatunnel.connectors.doris.sink.writer.RecordStream.endInput(RecordStream.java:41) ~[?:?]
worker2 | at org.apache.seatunnel.connectors.doris.sink.writer.DorisStreamLoad.stopLoad(DorisStreamLoad.java:224) ~[?:?]
worker2 | at org.apache.seatunnel.connectors.doris.sink.writer.DorisSinkWriter.flush(DorisSinkWriter.java:201) ~[?:?]
worker2 | at org.apache.seatunnel.connectors.doris.sink.writer.DorisSinkWriter.close(DorisSinkWriter.java:257) ~[?:?]
worker2 | at org.apache.seatunnel.api.sink.multitablesink.MultiTableSinkWriter.lambda$close$5(MultiTableSinkWriter.java:331) ~[seatunnel-starter.jar:2.3.10]
worker2 | at java.util.concurrent.ConcurrentHashMap.forEach(ConcurrentHashMap.java:1597) ~[?:1.8.0_342]
worker2 | at org.apache.seatunnel.api.sink.multitablesink.MultiTableSinkWriter.close(MultiTableSinkWriter.java:328) ~[seatunnel-starter.jar:2.3.10]
worker2 | at org.apache.seatunnel.engine.server.task.flow.SinkFlowLifeCycle.close(SinkFlowLifeCycle.java:174) ~[seatunnel-starter.jar:2.3.10]
worker2 | at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$close$5(SeaTunnelTask.java:330) ~[seatunnel-starter.jar:2.3.10]
worker2 | at org.apache.seatunnel.api.tracing.MDCConsumer.accept(MDCConsumer.java:43) ~[seatunnel-starter.jar:2.3.10]
worker2 | at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183) ~[?:1.8.0_342]
worker2 | at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384) ~[?:1.8.0_342]
worker2 | at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) ~[?:1.8.0_342]
worker2 | at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:290) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.ForkJoinPool.helpComplete(ForkJoinPool.java:1870) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.ForkJoinPool.externalHelpComplete(ForkJoinPool.java:2467) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.ForkJoinTask.externalAwaitDone(ForkJoinTask.java:324) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.ForkJoinTask.doInvoke(ForkJoinTask.java:405) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:734) ~[?:1.8.0_342]
worker2 | at java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:159) ~[?:1.8.0_342]
worker2 | at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:173) ~[?:1.8.0_342]
worker2 | at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233) ~[?:1.8.0_342]
worker2 | at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485) ~[?:1.8.0_342]
worker2 | at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:650) ~[?:1.8.0_342]
worker2 | at org.apache.seatunnel.api.tracing.MDCStream.forEach(MDCStream.java:88) ~[seatunnel-starter.jar:2.3.10]
worker2 | at org.apache.seatunnel.engine.server.task.SeaTunnelTask.close(SeaTunnelTask.java:327) ~[seatunnel-starter.jar:2.3.10]
worker2 | at org.apache.seatunnel.engine.server.TaskExecutionService$BlockingWorker.run(TaskExecutionService.java:711) ~[seatunnel-starter.jar:2.3.10]
worker2 | at org.apache.seatunnel.engine.server.TaskExecutionService$NamedTaskWrapper.run(TaskExecutionService.java:1019) ~[seatunnel-starter.jar:2.3.10]
worker2 | at org.apache.seatunnel.api.tracing.MDCRunnable.run(MDCRunnable.java:43) ~[seatunnel-starter.jar:2.3.10]
worker2 | at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_342]
worker2 | at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_342]
worker2 | at java.lang.Thread.run(Thread.java:750) [?:1.8.0_342]
worker2 | Caused by: org.apache.seatunnel.connectors.doris.exception.DorisConnectorException: ErrorCode:[Doris-01], ErrorDescription:[stream load error] - [NOT_FOUND]TStatus: errCode = 7, detailMessage = table not found, tableName=z_auth_role
```
### 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
Reproduce the Zeta-mode MySQL CDC to Doris 3.0+ configuration and begin with RecordBuffer.java, RecordStream.java, DorisStreamLoad.java, and DorisSinkWriter.java from the stack trace. Trace the CREATE_SCHEMA_WHEN_NOT_EXIST path for the generated z_auth_role table; done means initial CDC processing automatically creates the table and avoids the Doris table-not-found error.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, mysql
- Domain
- data-engineering, databases
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Needs clarification
- Newbie friendliness
- 45/100