[
https://issues.apache.org/jira/browse/FLINK-41012?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18127686#comment-18127686
]
zhanglu153 edited comment on FLINK-41012 at 10/16/26 5:53 AM:
--------------------------------------------------------------
The job never leaves the restart loop on its own; manual intervention (kill the
job, drop state, resubmit) is required.
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:
# 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.
Root cause (three stacked defects)
1. Non-atomic two-phase schema bookkeeping (root cause)
`SchemaCoordinator.applySchemaChange` writes the original schema to the
in-memory `SchemaManager` *before* applying to the external system:
{code:java}
// SchemaCoordinator.java
374: if (!SchemaUtils.isSchemaChangeEventRedundant(currentUpstreamSchema,
originalEvent)) {
375: schemaManager.applyOriginalSchemaChange(originalEvent); // ←
written first
376:
deducedSchemaChangeEvents.addAll(deduceEvolvedSchemaChanges(originalEvent));
...
438: private boolean applyAndUpdateEvolvedSchemaChange(SchemaChangeEvent
schemaChangeEvent) {
439: try {
440: metadataApplier.applySchemaChange(schemaChangeEvent); // ←
throws here
441: schemaManager.applyEvolvedSchemaChange(schemaChangeEvent); // ←
never reached
...
454: throw t; {code}
When L440 throws, the coordinator is left in a half-committed state:
originalSchemas contains the table, evolvedSchemas does not.
2. Redundancy check cannot distinguish half-committed from fully-applied
(lock-in point)
SchemaUtils.isSchemaChangeEventRedundant treats the mere presence of the
original schema as "already applied":
{code:java}
// SchemaUtils.java
359: createTableEvent -> {
360: // It has been applied if such table already exists
361: return latestSchema.isPresent(); // ← checks originalSchema only
362: }, {code}
3. Why the half-committed state survives restarts
With 0 completed checkpoints, SchemaRegistry.resetToCheckpoint returns early
and never clears the in-memory SchemaManager:
{code:java}
// SchemaRegistry.java
325: public final void resetToCheckpoint(long checkpointId, @Nullable byte[]
checkpointData) {
328: if (checkpointData == null) {
329: return; // ← memory state (incl. the half-committed one)
survives
330: } {code}
(If a checkpoint is taken during the crash loop, the poisoned state gets
persisted and restored instead — same outcome, even more insidious.)
Crash loop after recovery:
{code:java}
restart → Source re-emits CreateTableEvent
→ originalSchema present → judged redundant → "just skip it"
(DDL retry is skipped forever — FE recovery is never noticed)
→ response contains {table=null} via
schemaManager.getLatestEvolvedSchema(tableId).orElse(null) //
L421-422
→ SchemaOperator: evolvedSchemaMap.putAll(...) pollutes the
operator-local cache with a null value // L191
→ first DataChangeEvent → coerceDataRecord fails → crash // L217-225
→ job fails → restart → goto 1 {code}
The key observation: after the initial failure the coordinator never talks to
the sink again, so whether the external system recovered is irrelevant — the
loop has no exit.
Expected behaviour
After the external system recovers, the job should apply the pending
CreateTableEvent and converge to RUNNING. A transient sink failure during
initial table creation must not permanently corrupt coordinator state.
was (Author: JIRAUSER298392):
The job never leaves the restart loop on its own; manual intervention (kill the
job, drop state, resubmit) is required.
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:
# 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.
Root cause (three stacked defects)
1. Non-atomic two-phase schema bookkeeping (root cause)
`SchemaCoordinator.applySchemaChange` writes the original schema to the
in-memory `SchemaManager` *before* applying to the external system:
{code:java}
// SchemaCoordinator.java
374: if (!SchemaUtils.isSchemaChangeEventRedundant(currentUpstreamSchema,
originalEvent)) {
375: schemaManager.applyOriginalSchemaChange(originalEvent); // ←
written first
376:
deducedSchemaChangeEvents.addAll(deduceEvolvedSchemaChanges(originalEvent));
...
438: private boolean applyAndUpdateEvolvedSchemaChange(SchemaChangeEvent
schemaChangeEvent) {
439: try {
440: metadataApplier.applySchemaChange(schemaChangeEvent); // ←
throws here
441: schemaManager.applyEvolvedSchemaChange(schemaChangeEvent); // ←
never reached
...
454: throw t; {code}
When L440 throws, the coordinator is left in a half-committed state:
originalSchemas contains the table, evolvedSchemas does not.
2. Redundancy check cannot distinguish half-committed from fully-applied
(lock-in point)
SchemaUtils.isSchemaChangeEventRedundant treats the mere presence of the
original schema as "already applied":
{code:java}
// SchemaUtils.java
359: createTableEvent -> {
360: // It has been applied if such table already exists
361: return latestSchema.isPresent(); // ← checks originalSchema only
362: }, {code}
3. Why the half-committed state survives restarts
With 0 completed checkpoints, SchemaRegistry.resetToCheckpoint returns early
and never clears the in-memory SchemaManager:
{code:java}
// SchemaRegistry.java
325: public final void resetToCheckpoint(long checkpointId, @Nullable byte[]
checkpointData) {
328: if (checkpointData == null) {
329: return; // ← memory state (incl. the half-committed one)
survives
330: } {code}
(If a checkpoint is taken during the crash loop, the poisoned state gets
persisted and restored instead — same outcome, even more insidious.)
Crash loop after recovery:
{code:java}
restart → Source re-emits CreateTableEvent
→ originalSchema present → judged redundant → "just skip it"
(DDL retry is skipped forever — FE recovery is never noticed)
→ response contains {table=null} via
schemaManager.getLatestEvolvedSchema(tableId).orElse(null) //
L421-422
→ SchemaOperator: evolvedSchemaMap.putAll(...) pollutes the
operator-local cache with a null value // L191
→ first DataChangeEvent → coerceDataRecord fails → crash // L217-225
→ job fails → restart → goto 1 {code}
The key observation: after the initial failure the coordinator never talks to
the sink again, so whether the external system recovered is irrelevant — the
loop has no exit.
> 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)
