[Bug] Failed when a table that was not included in the source changed its schema.

[lvyanquan]

Search before asking

• I searched in the issues and found nothing similar.

Flink version

1.18

Flink CDC version

3.0

Database and its version

Mysql

Minimal reproduce step

run pipeline job with the following file, capture only one inventory.geom table:

################################################################################
# Description: Sync MySQL one table to Values
################################################################################
source:
  type: mysql
  hostname: localhost
  port: 3306
  username: root
  password: 123456
  tables: inventory.geom
  server-id: 5400-5404
  server-time-zone: Asia/Shanghai

sink:
  type: values

pipeline:
  name: Sync MySQL Database to Values
  parallelism: 2

Then change a table schema of other tables, like adding a new column for inventory.orders.

What did you expect to see?

Nothing happen.

What did you see instead?

Got error.

Caused by: java.lang.IllegalArgumentException: Unable to apply SchemaChangeEvent for table "inventory.orders" without existing schema
	at com.ververica.cdc.common.utils.Preconditions.checkArgument(Preconditions.java:128) ~[classes/:?]
	at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaManager.applySchemaChange(SchemaManager.java:105) ~[classes/:?]
	at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistryRequestHandler.handleSchemaChangeRequest(SchemaRegistryRequestHandler.java:102) ~[classes/:?]
	at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistry.handleCoordinationRequest(SchemaRegistry.java:157) ~[classes/:?]
	at org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler.deliverCoordinationRequestToCoordinator(DefaultOperatorCoordinatorHandler.java:143) ~[flink-runtime-1.18.0.jar:1.18.0]

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!