[FLINK-37278] Optimize regular schema evolution topology's performance

[yuxiqian]

This closes FLINK-37278.

Currently, regular SE topology uses the following process to drain existing DataChangeEvents in the pipeline:

1. SchemaOperator ("client") emits FlushEvent to downstream.
2. The "client" keeps polling the SchemaCoordinator ("server") with 1-second interval.
3. The "server" rejects all requests from clients until it has collected enough FlushSuccessEvent notifications from Sink.

As a result, all schema change requests will took at least 1 second to finish, after at least one polling interval.

---

This PR replaces the polling code with maintaining a pending schema change request queue, where SchemaCoordinator could manage all pending clients and effectively blocking them from handling upstream events. Schema evolution process could start immediately after FlushSuccessEvent got reported, needless to wait for polling requests from clients.

---

With this change, time consumption of testRegularTablesSourceInMultipleParallelism test case has been reduced from ~6 minutes to ~50 seconds.

[yuxiqian]

Would @hiliuxg like to take a look?

[Shawn-Hx]

It seems that SchemaChangeResponse#ResponseCode can only be SUCCESS now. Can we remove SchemaChangeResponse#ResponseCode and simplify the logic in SchemaOperator#handleSchemaChangeEvent ?

[yuxiqian]

Thanks for Shawn's kindly review, comments addressed.

[Shawn-Hx]

LGTM

[gongzexin]

It seems that SchemaChangeResponse#ResponseCode can only be SUCCESS now. Can we remove SchemaChangeResponse#ResponseCode and simplify the logic in SchemaOperator#handleSchemaChangeEvent ?

@yuxiqian @Shawn-Hx
Hi, Have you noticed that during fault tolerance, the same table will be flushed multiple times (related to the task parallelism).So I think SchemaChangeResponse#ResponseCode#DUPLICATE should not be deleted, but it should be strengthened.

[yuxiqian]

It seems that SchemaChangeResponse#ResponseCode can only be SUCCESS now. Can we remove SchemaChangeResponse#ResponseCode and simplify the logic in SchemaOperator#handleSchemaChangeEvent ?

@yuxiqian @Shawn-Hx Hi, Have you noticed that during fault tolerance, the same table will be flushed multiple times (related to the task parallelism).So I think SchemaChangeResponse#ResponseCode#DUPLICATE should not be deleted, but it should be strengthened.

Thanks for @gongzexin's report. IIUC, the root cause of this problem is PreTransformOperator invokes getUnionListState to store persistent schemas, all subTasks of SchemaOperators will obtain the same set of table schemas when restoring from state, and SchemaCoordinator is expected to receive $N$ duplicate requests ($N$ = parallelism). Worse still, UnionListState will block the checkpointing process when some subTasks have entered the FINISHED state (FLINK-37368).

I wonder if we can handle it in another PR, and focus on modifying the schema evolution request queueing logic here?

[lvyanquan]

LGTM.

[lvyanquan]

Hi, @leonardBang @ruanhang1993, could you take a look at this?

[nihaoya2025]

@yuxiqian I have a table that I want to delete and resynchronize with both full and incremental data. The schema recorded by the Schema Manager will not be automatically deleted when the table is deleted, skipping the new table creation statement. What should I do

 // For redundant schema change events (possibly coming from duplicate emitted
        // CreateTableEvents in snapshot stage), we just skip them.
        if (!SchemaUtils.isSchemaChangeEventRedundant(currentUpstreamSchema, originalEvent)) {
            schemaManager.applyOriginalSchemaChange(originalEvent);
            deducedSchemaChangeEvents.addAll(deduceEvolvedSchemaChanges(originalEvent));
        }