[FLINK-37979] Remove obsolete MySQL CDC snapshot split assigner code

[morozov]

There is a redundant condition and a misleading comment in MySqlSnapshotSplitAssigner#captureNewlyAddedTables() that complicate understanding of the code:

flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

Lines 280 to 285 in 1646715

	if (AssignerStatus.isAssigningFinished(assignerStatus)) {
	// start the newly added tables process under binlog reading phase
	LOG.info(
	"Found newly added tables, start capture newly added tables process under binlog reading phase");
	this.startAssignNewlyAddedTables();
	}

By the time when the above condition is being evaluated, an identical condition has been already evaluated earlier in the code:

flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

Lines 226 to 228 in 1646715

	if (sourceConfig.isScanNewlyAddedTableEnabled()
	&& !sourceConfig.getStartupOptions().isSnapshotOnly()
	&& AssignerStatus.isAssigningFinished(assignerStatus)) {

Therefore, the second evaluation will always result in true and thus is redundant.

Code Change History

Prior to #3519, table discovery would happen in any assigner status, so the condition in the end of the method made sense. Together with adding this condition in the beginning of the method, #3519 should have removed it from the end.

What Exactly is Obsolete

1. The comment:

   flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

   Lines 274 to 275 in 1646715

   	// if job is still in snapshot reading phase, directly add all newly added
   	// tables

   
   Since [FLINK-35859][cdc-connector] Fix: The assigner is not ready to offer finished split information, this should not be called #3519, if the job is still in the snapshot reading phase, this code won't be executed. This comment should be removed as misleading.
2. The second if (AssignerStatus.isAssigningFinished(assignerStatus)) statement. While the captureNewlyAddedTables() method is running, the assigner status cannot be modified by a different thread because this code is only executed as part of MySqlSourceEnumerator#open(), i.e. the source in its initialization phase.

[lvyanquan]

+1. Thanks for your detailed analysis.