zhanglu153 created FLINK-41012:
----------------------------------
Summary: Schema change apply failure leads to unrecoverable crash
loop: redundant skip + null evolved schema leak after job restart
Key: FLINK-41012
URL: https://issues.apache.org/jira/browse/FLINK-41012
Project: Flink
Issue Type: Bug
Components: Flink CDC
Affects Versions: cdc-3.6.0
Environment: ## Environment
- Flink CDC: 3.6.0
- Flink: 1.20.4, YARN deployment
- Pipeline: MySQL → Doris, `schema.change.behavior: evolve`, checkpoint
interval 600s
- The reproduction requires **0 completed checkpoints** (default restart
strategy keeps retrying)
## Reproduction
Reproduced reliably (twice) with a timing-controlled fault injection:
1. Submit a fresh pipeline job (MySQL → Doris, no existing sink tables).
2. Right after both subtasks reach RUNNING and before the first
`CreateTableEvent`
is applied (~20s window), inject a transient HTTP failure on the Doris FE
(we used Chaos Mesh HTTPChaos: abort all requests on port 8030 — TCP stays
connectable, HTTP requests fail).
3. The first `applyCreateTableEvent` throws
`DorisRuntimeException: No Doris FE is available` from
`SchemaCoordinator.applyAndUpdateEvolvedSchemaChange` (L440). Job fails and
restarts.
4. Remove the HTTPChaos (FE immediately healthy again, Doris never restarted,
job configuration untouched).
5. Job keeps restarting forever. Observed for 6+ minutes after external
recovery:
crash count and null-schema log lines keep growing in lockstep.
Reporter: zhanglu153
When a `CreateTableEvent` fails to apply to the external system (e.g. the sink
database is temporarily unreachable) and the job restarts before any checkpoint
completes, the job enters an unrecoverable crash loop. Even after the external
failure is fully resolved, every restart fails within ~1 second with:
{code:java}
java.lang.IllegalStateException: Unable to coerce data record from
app_db.orders (schema: columns={`id` INT NOT NULL,`price` DECIMAL(10, 2) NOT
NULL}, primaryKeys=id, options=()) to app_db.orders (schema: null) at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator.lambda$handleDataChangeEvent$1(SchemaOperator.java:220)
at java.base/java.util.Optional.orElseThrow(Optional.java:408) at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator.handleDataChangeEvent(SchemaOperator.java:217)
at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaOperator.processElement(SchemaOperator.java:155)
at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at
org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:310)
at
org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110)
at
org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:101)
at
org.apache.flink.cdc.connectors.mysql.source.reader.MySqlRecordEmitter$OutputCollector.collect(MySqlRecordEmitter.java:169)
at
java.base/java.util.Collections$SingletonList.forEach(Collections.java:4856) at
org.apache.flink.cdc.debezium.event.DebeziumEventDeserializationSchema.deserialize(DebeziumEventDeserializationSchema.java:106)
at
org.apache.flink.cdc.connectors.mysql.source.reader.MySqlRecordEmitter.emitElement(MySqlRecordEmitter.java:135)
at
org.apache.flink.cdc.connectors.mysql.source.reader.MySqlRecordEmitter.processElement(MySqlRecordEmitter.java:108)
at
org.apache.flink.cdc.connectors.mysql.source.reader.MySqlPipelineRecordEmitter.processElement(MySqlPipelineRecordEmitter.java:178)
at
org.apache.flink.cdc.connectors.mysql.source.reader.MySqlRecordEmitter.emitRecord(MySqlRecordEmitter.java:80)
at
org.apache.flink.cdc.connectors.mysql.source.reader.MySqlRecordEmitter.emitRecord(MySqlRecordEmitter.java:47)
at
org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:203)
at
org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:443)
at
org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
at
org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:638)
at
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:973)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:917)
at
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:949)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763) at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:575) at
java.base/java.lang.Thread.run(Thread.java:834) {code}
The most critical thing is JM's task error logs.:
{code:java}
2026-10-15 17:20:21,442 WARN org.apache.doris.flink.sink.BackendUtil
[] - Failed to connect to
host:http://cdb-falco-fenode.lytest.svc.cluster.local:8030java.net.SocketException:
Unexpected end of file from server at
java.base/sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:862)
~[?:?] at
java.base/sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:689) ~[?:?] at
java.base/sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:859)
~[?:?] at
java.base/sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:689) ~[?:?] at
java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1615)
~[?:?] at
java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1520)
~[?:?] at
java.base/java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:527)
~[?:?] at
org.apache.doris.flink.sink.BackendUtil.tryHttpConnection(BackendUtil.java:108)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.rest.RestService.randomEndpoint(RestService.java:287)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.buildHttpPost(SchemaChangeManager.java:283)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.sendHttpPostRequest(SchemaChangeManager.java:331)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.checkDatabaseExists(SchemaChangeManager.java:326)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.createTable(SchemaChangeManager.java:77)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applyCreateTableEvent(DorisMetadataApplier.java:178)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.lambda$applySchemaChange$2(DorisMetadataApplier.java:126)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:59)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applySchemaChange(DorisMetadataApplier.java:115)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:440)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:409)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:250)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
[?:?] at
java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?] at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[?:?] at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[?:?] at java.base/java.lang.Thread.run(Thread.java:834) [?:?]2026-10-15
17:20:21,445 ERROR
org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry [] - An
exception was triggered from Schema change applying task. Job will fail
now.org.apache.flink.util.FlinkRuntimeException: Failed to apply schema change
event. at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:252)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
[?:?] at
java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?] at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[?:?] at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[?:?] at java.base/java.lang.Thread.run(Thread.java:834) [?:?]Caused by:
org.apache.flink.cdc.common.exceptions.SchemaEvolveException:
org.apache.doris.flink.exception.DorisRuntimeException: No Doris FE is
available, please check configuration or cluster status. at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applyCreateTableEvent(DorisMetadataApplier.java:180)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.lambda$applySchemaChange$2(DorisMetadataApplier.java:126)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:59)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applySchemaChange(DorisMetadataApplier.java:115)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:440)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:409)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:250)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] ... 5
moreCaused by: org.apache.doris.flink.exception.DorisRuntimeException: No Doris
FE is available, please check configuration or cluster status. at
org.apache.doris.flink.rest.RestService.randomEndpoint(RestService.java:291)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.buildHttpPost(SchemaChangeManager.java:283)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.sendHttpPostRequest(SchemaChangeManager.java:331)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.checkDatabaseExists(SchemaChangeManager.java:326)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.createTable(SchemaChangeManager.java:77)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applyCreateTableEvent(DorisMetadataApplier.java:178)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.lambda$applySchemaChange$2(DorisMetadataApplier.java:126)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:59)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applySchemaChange(DorisMetadataApplier.java:115)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:440)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:409)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:250)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] ... 5
more2026-10-15 17:20:21,446 ERROR
org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry [] -
Uncaught exception in the Schema Registry (SchemaOperator) event loop for
Schema change applying task.org.apache.flink.util.FlinkRuntimeException: Failed
to apply schema change event. at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:252)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
[?:?] at
java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?] at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[?:?] at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[?:?] at java.base/java.lang.Thread.run(Thread.java:834) [?:?]Caused by:
org.apache.flink.cdc.common.exceptions.SchemaEvolveException:
org.apache.doris.flink.exception.DorisRuntimeException: No Doris FE is
available, please check configuration or cluster status. at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applyCreateTableEvent(DorisMetadataApplier.java:180)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.lambda$applySchemaChange$2(DorisMetadataApplier.java:126)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:59)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applySchemaChange(DorisMetadataApplier.java:115)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:440)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:409)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:250)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] ... 5
moreCaused by: org.apache.doris.flink.exception.DorisRuntimeException: No Doris
FE is available, please check configuration or cluster status. at
org.apache.doris.flink.rest.RestService.randomEndpoint(RestService.java:291)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.buildHttpPost(SchemaChangeManager.java:283)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.sendHttpPostRequest(SchemaChangeManager.java:331)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.checkDatabaseExists(SchemaChangeManager.java:326)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.createTable(SchemaChangeManager.java:77)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applyCreateTableEvent(DorisMetadataApplier.java:178)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.lambda$applySchemaChange$2(DorisMetadataApplier.java:126)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:59)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applySchemaChange(DorisMetadataApplier.java:115)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:440)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:409)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:250)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] ... 5
more2026-10-15 17:20:21,446 ERROR
org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry [] -
Current schema manager state: Schema Manager -54527316: original schema
map: - table app_db.orders: {app_db.orders={0=columns={`id` INT NOT
NULL,`price` DECIMAL(10, 2) NOT NULL}, primaryKeys=id, options=()}}
evolved schema map:
2026-10-15 17:20:21,461 INFO org.apache.flink.runtime.jobmaster.JobMaster
[] - Trying to recover from a global
failure.org.apache.flink.util.FlinkException: Global failure triggered by
OperatorCoordinator for 'Source: Flink CDC Event Source: mysql ->
SchemaOperator -> PrePartition' (operator 9899a42c64d67ef3172b7e3be3c1bbb9).
at
org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:651)
~[flink-dist-1.20.4-HDP-26.10.1.jar:1.20.4-HDP-26.10.1] at
org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry.failJob(SchemaRegistry.java:397)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:252)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
~[?:?] at
java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?] at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
~[?:?] at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
~[?:?] at java.base/java.lang.Thread.run(Thread.java:834) ~[?:?]Caused by:
org.apache.flink.util.FlinkRuntimeException: Failed to apply schema change
event. ... 6 moreCaused by:
org.apache.flink.cdc.common.exceptions.SchemaEvolveException:
org.apache.doris.flink.exception.DorisRuntimeException: No Doris FE is
available, please check configuration or cluster status. at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applyCreateTableEvent(DorisMetadataApplier.java:180)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.lambda$applySchemaChange$2(DorisMetadataApplier.java:126)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:59)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applySchemaChange(DorisMetadataApplier.java:115)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:440)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:409)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:250)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] ... 5
moreCaused by: org.apache.doris.flink.exception.DorisRuntimeException: No Doris
FE is available, please check configuration or cluster status. at
org.apache.doris.flink.rest.RestService.randomEndpoint(RestService.java:291)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.buildHttpPost(SchemaChangeManager.java:283)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.sendHttpPostRequest(SchemaChangeManager.java:331)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.checkDatabaseExists(SchemaChangeManager.java:326)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.doris.flink.sink.schema.SchemaChangeManager.createTable(SchemaChangeManager.java:77)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applyCreateTableEvent(DorisMetadataApplier.java:178)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.lambda$applySchemaChange$2(DorisMetadataApplier.java:126)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:59)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.connectors.doris.sink.DorisMetadataApplier.applySchemaChange(DorisMetadataApplier.java:115)
~[flink-cdc-pipeline-connector-doris-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2]
at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:440)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:409)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] at
org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:250)
~[flink-cdc-dist-3.6.0-HDP-26.06.2.jar:3.6.0-HDP-26.06.2] ... 5 more
{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
