[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.

[zaki-zhao]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

Zeta模式下MySQL CDC到Doris无法自动建表，Doris版本3.0+，且报错

SeaTunnel Version

2.3.10

SeaTunnel Config

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

-

Error Exception

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

• I agree to follow this project's Code of Conduct

[davidzollo]

I use doris 2.x and SeaTunnel can create table normally when using CREATE_SCHEMA_WHEN_NOT_EXIST
Please provide the clear version of Doris @zaki-zhao
Is there andbody who'd like to find out the root cause? please leave a message

[zaki-zhao]

I use doris 2.x and SeaTunnel can create table normally when using CREATE_SCHEMA_WHEN_NOT_EXIST Please provide the clear version of Doris @zaki-zhao Is there andbody who'd like to find out the root cause? please leave a message

doris version 3.0.3