[jira] [Created] (FLINK-39696) Distributed SchemaOperator emits FlushEvent with schema operator subtaskId instead of upstream sourcePartition

["Ran Tao (Jira)" <ji...@apache.org>]

Ran Tao created FLINK-39696:
-------------------------------

             Summary: Distributed SchemaOperator emits FlushEvent with schema operator subtaskId instead of upstream sourcePartition
                 Key: FLINK-39696
                 URL: https://issues.apache.org/jira/browse/FLINK-39696
             Project: Flink
          Issue Type: Bug
          Components: Flink CDC
            Reporter: Ran Tao


In the distributed schema evolution topology, the source identity used by the schema-change path and the flush path is inconsistent.

{*}Current behavior{*}:
1. DistributedPrePartitionOperator preserves the original upstream source partition when it broadcasts a schema change event.
2. Distributed SchemaOperator forwards that identity to the coordinator via {*}SchemaChangeRequest(sourcePartition, sinkSubTaskId, schemaChangeEvent){*}.
3. SchemaCoordinator also deduplicates distributed schema change events by (sourcePartition, schemaChangeEvent).
4. However, when SchemaOperator emits the downstream FlushEvent, it currently uses the SchemaOperator subtask id instead of the original source partition.

As a result, the same distributed schema change is identified by different keys in different parts of the pipeline:
 * SchemaChangeRequest / coordinator deduplication use the original source partition
 * FlushEvent / FlushSuccessEvent use the SchemaOperator subtask id

This makes the flush lineage inconsistent with the schema-change lineage in distributed topology, and duplicated broadcast branches of one upstream schema change cannot be consistently aligned by the original source partition.

{*}Expected behavior{*}:
FlushEvent in distributed SchemaOperator should carry the original upstream source partition, i.e. the same source identity already used in SchemaChangeRequest.

{*}Suggested fix{*}:
In {*}org.apache.flink.cdc.runtime.operators.schema.distributed.SchemaOperator#requestSchemaChange{*}, construct FlushEvent with schemaChangeRequest.getSourceSubTaskId() instead of the current SchemaOperator subtask id.

A simple regression test can reproduce the issue:
 * source partition = 0
 * distributed SchemaOperator subtask = 1
 * process a CreateTableEvent
 * the first emitted FlushEvent should use 0, not 1



--
This message was sent by Atlassian Jira
(v8.20.10#820010)