[FLINK-36701][cdc-runtime] Add steps to get and emit schemaManager's latest evolvedSchema when SinkDataWriterOperator handles FlushEvent

[Jzjsnow]

Currently, directly after a failover, when the pipeline first handles a schema change event (e.g. AddColumnEvent) and then a DataChangeEvent, it may cause the job to fail again as sink has repeatedly applied that schema change.

The issue is revealed as follows:

I add steps to get and emit schemaManager's latest evolvedSchema when SinkDataWriterOperator handles FlushEvent. If the operator doesn't have a local cache of the schema when handling the flushevent, it will request Schema Manager's latest evolved schema. At this point the evolved schema is the same with the sink table's, as the schema change event hasn't been applied to the evolved schema by the master yet.

Now the new process is changed as follows:

[yuxiqian]

Thanks for @Jzjsnow's clear and detailed diagram!

Just noticed that we still lack failover tests for pipeline jobs. Could we add some recovering tests like MySqlSourceITCase#testMySqlParallelSource and trigger TM/JM Failures in different phases? flink-cdc-composer unit tests might be a place for it.

[Jzjsnow]

Thanks for @Jzjsnow's clear and detailed diagram!

Just noticed that we still lack failover tests for pipeline jobs. Could we add some recovering tests like MySqlSourceITCase#testMySqlParallelSource and trigger TM/JM Failures in different phases? flink-cdc-composer unit tests might be a place for it.

@yuxiqian Thanks for the suggestion, I have added some tests to the flink-cdc-runtime unit test for testing DataSinkOperator's handling of schema change events, which was lacking. Among them, DataSinkOperatorWithSchemaEvolveTest#testSchemaChangeEventAfterFailover is used to test the underlying schema change process after failure recovery, which is applicable to the scenario of restarting after jm/tm failure.

[Jzjsnow]

Here it looks like we have once again encountered a OceanBaseMySQLModeITCase test failure similar to the one in #3712. Any ideas on how to fix this?

[yuxiqian]

It's worrying to notice similar OceanBase test case failures here, but seems irrelevant to this PR. Will investigate this.

[yuxiqian]

@Jzjsnow Should be fixed now... please rebase to master branch.

[Jzjsnow]

@yuxiqian Thanks for the quick fix, now we've rebased to the master branch.

[Jzjsnow]

The Source E2E Tests#OceanBaseE2eITCase test did not pass in this round of CIs, but it is not relevant to this PR. It seems that the 1 minute timeout set in the method checkResultWithTimeout (timeout=60000L) is a bit too short, so that the row id=111 in the sink table that should have been deleted has not been synchronized yet.

Here is the error log:

2025-01-03T07:38:53.8553280Z 1254259 [main] ERROR org.apache.flink.cdc.connectors.tests.OceanBaseE2eITCase -
2025-01-03T07:38:53.8553917Z --------------------------------------------------------------------------------
2025-01-03T07:38:53.8554680Z Test testOceanBaseCDCflinkVersion: 1.19.1 failed with:
2025-01-03T07:38:53.8555684Z array lengths differed, expected.length=10 actual.length=11; arrays first differed at element [10]; expected: but was:<111,scooter,Big 2-wheel scooter ,5.18,null,null>
2025-01-03T07:38:53.8556858Z at org.junit.internal.ComparisonCriteria.arrayEquals(ComparisonCriteria.java:89)
2025-01-03T07:38:53.8557558Z at org.junit.internal.ComparisonCriteria.arrayEquals(ComparisonCriteria.java:28)
2025-01-03T07:38:53.8558017Z at org.junit.Assert.internalArrayEquals(Assert.java:534)
2025-01-03T07:38:53.8558492Z at org.junit.Assert.assertArrayEquals(Assert.java:285)
2025-01-03T07:38:53.8558895Z at org.junit.Assert.assertArrayEquals(Assert.java:300)
2025-01-03T07:38:53.8559341Z at org.apache.flink.cdc.common.test.utils.JdbcProxy.checkResult(JdbcProxy.java:70)
2025-01-03T07:38:53.8559996Z at org.apache.flink.cdc.common.test.utils.JdbcProxy.checkResultWithTimeout(JdbcProxy.java:93)
2025-01-03T07:38:53.8560819Z at org.apache.flink.cdc.connectors.tests.OceanBaseE2eITCase.testOceanBaseCDC(OceanBaseE2eITCase.java:179)

[yuxiqian]

Thanks for @Jzjsnow's great work, left some minor comments.

[yuxiqian]

I'm wondering if we may need to trace FlushEvent more specifically, about what type of schema change event it was caused by.

If so, we may store a SchemaChangeEventType enum value for extensibility.

If it turns out to be unnecessary, at least we can use boolean to avoid [un]boxing.

[yuxiqian]

Maybe this change can be split into an individual commit to keep commit history accurate.