[jira] [Updated] (FLINK-38334) MySQL CDC source may get stuck in the INITIAL_ASSIGNING state

["Sergei Morozov (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-38334?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Sergei Morozov updated FLINK-38334:
-----------------------------------
    Description: 
The logic of filtering out irrelevant snapshot splits on the enumerator and the source reader is inconsistent:
 # The enumerator filters them out only if the assigner is in the INITIAL_ASSIGNING_FINISHED or the NEWLY_ADDED_ASSIGNING_FINISHED state ([source|https://github.com/apache/flink-cdc/blob/release-3.2.0/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java#L219]).
 # The source reader filters them out regardless of the assigner state.

As a result, if a table is excluded from the source configuration during the snapshot of that table, the source reader will drop its splits while the enumerator will expect them to be processed by the source reader and reported back. As a result, the source gets stuck in the snapshotting phase and never transitions to streaming.

h5. Steps to reproduce

# Prepare tables {{A}} and {{B}} to be captured by the source. The number of rows in {{B}} is irrelevant, table {{A}} needs to have enough rows for its snapshot to take a couple of minutes (100 rows should be enough given the configuration below).
# Make sure the source has the following configuration parameters:
## {{scan.snapshot.fetch.size: 2}}. This will guarantee that the snapshot of table {{A}} will take long enough ({{2}} is the minimum allowed value).
# Include tables {{A}} and {{B}} into the source's configuration.
# Start the job and wait until it starts snapshotting {{A}}.
# Stop the job.
# Exclude table {{A}} from the configuration.
# Start the job.
# Observe that the source reader drops the splits that belong to {{A}} (should log an "is skipping split" {{INFO}} message for each). At the same time, the enumerator still expects them to be finished and reported.
# Make changes in table {{B}} (e.g. insert a row)
# Observe that this change never makes its way downstream.

  was:
The logic of filtering out irrelevant snapshot splits on the enumerator and the source reader is inconsistent:
 # The enumerator filters them out only if the assigner is in the INITIAL_ASSIGNING_FINISHED or the NEWLY_ADDED_ASSIGNING_FINISHED state ([source|https://github.com/apache/flink-cdc/blob/release-3.2.0/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java#L219]).
 # The source reader filters them out regardless of the assigner state.

As a result, if a table is excluded from the source configuration during the snapshot of that table, the source reader will drop its splits while the enumerator will expect them to be processed by the source reader and reported back. As a result, the source gets stuck in the snapshotting phase and never transitions to streaming.


> MySQL CDC source may get stuck in the INITIAL_ASSIGNING state
> -------------------------------------------------------------
>
>                 Key: FLINK-38334
>                 URL: https://issues.apache.org/jira/browse/FLINK-38334
>             Project: Flink
>          Issue Type: Bug
>          Components: Flink CDC
>    Affects Versions: cdc-3.2.0
>            Reporter: Sergei Morozov
>            Priority: Major
>
> The logic of filtering out irrelevant snapshot splits on the enumerator and the source reader is inconsistent:
>  # The enumerator filters them out only if the assigner is in the INITIAL_ASSIGNING_FINISHED or the NEWLY_ADDED_ASSIGNING_FINISHED state ([source|https://github.com/apache/flink-cdc/blob/release-3.2.0/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java#L219]).
>  # The source reader filters them out regardless of the assigner state.
> As a result, if a table is excluded from the source configuration during the snapshot of that table, the source reader will drop its splits while the enumerator will expect them to be processed by the source reader and reported back. As a result, the source gets stuck in the snapshotting phase and never transitions to streaming.
> h5. Steps to reproduce
> # Prepare tables {{A}} and {{B}} to be captured by the source. The number of rows in {{B}} is irrelevant, table {{A}} needs to have enough rows for its snapshot to take a couple of minutes (100 rows should be enough given the configuration below).
> # Make sure the source has the following configuration parameters:
> ## {{scan.snapshot.fetch.size: 2}}. This will guarantee that the snapshot of table {{A}} will take long enough ({{2}} is the minimum allowed value).
> # Include tables {{A}} and {{B}} into the source's configuration.
> # Start the job and wait until it starts snapshotting {{A}}.
> # Stop the job.
> # Exclude table {{A}} from the configuration.
> # Start the job.
> # Observe that the source reader drops the splits that belong to {{A}} (should log an "is skipping split" {{INFO}} message for each). At the same time, the enumerator still expects them to be finished and reported.
> # Make changes in table {{B}} (e.g. insert a row)
> # Observe that this change never makes its way downstream.



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