[
https://issues.apache.org/jira/browse/FLINK-41012?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18127757#comment-18127757
]
zhanglu153 commented on FLINK-41012:
------------------------------------
## Fix
A `CreateTableEvent` is now considered redundant **only if the evolved schema
has also been applied for all routed sink tables**. A half-committed state
re-enters the full path so the DDL is retried on every restart:
{code:java}
boolean isRedundant =
SchemaUtils.isSchemaChangeEventRedundant(currentUpstreamSchema,
originalEvent);// A CreateTableEvent is only truly redundant if the evolved
schema has also been
// applied to all routed sink tables. A half-committed state (original schema
// present but evolved schema missing, e.g. caused by a failed external apply
// followed by a job restart) must re-enter the full path to retry the DDL;
// otherwise the job crashes on the next DataChangeEvent with
// "Unable to coerce data record ... (schema: null)" and can never recover.
if (isRedundant && originalEvent instanceof CreateTableEvent) {
for (TableId sinkTableId : router.route(originalTableId)) {
if (!schemaManager.evolvedSchemaExists(sinkTableId)) {
isRedundant = false;
LOG.warn(
"Detected half-committed schema change for table {}:
original schema "
+ "exists but evolved schema for sink table {} is
missing. "
+ "The schema change will be re-applied.",
originalTableId,
sinkTableId);
break;
}
}
}if (!isRedundant) {
schemaManager.applyOriginalSchemaChange(originalEvent);
deducedSchemaChangeEvents.addAll(deduceEvolvedSchemaChanges(originalEvent));
} else {
// skip (both schemas are guaranteed to exist here)
}
{code}
How it works:SchemaCoordinator.applySchemaChange registers the original schema
before applying to the external system. If the external apply fails, the
coordinator is left half-committed: original schema present, evolved schema
missing. This state survives restarts (no checkpoint clears it).On restart, the
source re-emits the CreateTableEvent. Previously, the redundancy check only
looked at the original schema, judged the event redundant, and skipped it
forever — so the DDL was never retried, and the leaked \{table=null} in the
response crashed every subsequent DataChangeEvent.With the guard, the
half-committed state is detected (evolved schema missing for any routed sink
table), the event is re-applied, and once the external system is back the DDL
succeeds — the evolved schema is registered, the operator cache is refreshed
with the real schema, and the job converges to RUNNING.Key property: the fix
does not prevent applySchemaChange from failing while the sink is down (that is
correct behaviour). It removes the "never retry again" lock-in: every restart
honestly retries the DDL, and the first restart after the external recovery
succeeds. Normal duplicate events (snapshot phase) are unaffected — once fully
applied, both schemas exist and the guard is a no-op.
> 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: # 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)
> Reporter: zhanglu153
> Priority: Critical
> Labels: Flink-CDC
>
> 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)
