[Improve][Zeta] Filter tasks and pipelines by state

[dybyte]

Purpose of this pull request

• Exclude tasks and pipelines in CLOSED/FINISHED state from checkpoint, statistics, and trigger operations to improve system stability and performance.
• Updated logic in CheckpointCoordinator.java and PhysicalPlanGenerator.java to filter out tasks and pipelines no longer active.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added tests ensuring CLOSED/FINISHED tasks and pipelines are properly excluded from processing.

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[Hisoka-X]

Could you add a e2e test case with savepoint and restore for cdc job? This is a very important and useful change. Thanks @dybyte !

[dybyte]

Could you add a e2e test case with savepoint and restore for cdc job?

seatunnel/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithBinlogDeleteIT.java

Lines 143 to 216 in a0aa0ce

	public void testRestoreTaskWhenBinlogDelete(TestContainer container)
	throws InterruptedException, IOException {
	// Clear related content to ensure that multiple operations are not affected
	clearTable(MYSQL_DATABASE, SINK_TABLE);
	// execute task
	Long jobId = JobIdGenerator.newJobId();
	CompletableFuture.supplyAsync(
	() -> {
	try {
	return container.executeJob(
	"/mysqlcdc_to_mysql_with_binlog_delete.conf",
	String.valueOf(jobId));
	} catch (Exception e) {
	log.error("Commit task exception :" + e.getMessage());
	throw new RuntimeException(e);
	}
	});
	// wait for data written to sink
	await().atMost(60000, TimeUnit.MILLISECONDS)
	.untilAsserted(
	() -> {
	Assertions.assertIterableEquals(
	query(getSourceQuerySQL(MYSQL_DATABASE, SOURCE_TABLE)),
	query(getSinkQuerySQL(MYSQL_DATABASE, SINK_TABLE)));
	});
	// flush binary logs
	executeSql("flush binary logs");
	// wait a moment for binlog heartbeat event
	TimeUnit.SECONDS.sleep(60);
	// pause task
	Assertions.assertEquals(0, container.savepointJob(String.valueOf(jobId)).getExitCode());
	// purge binary logs
	List<List<Object>> masterStatus = query("show master status");
	String binlogName = masterStatus.get(0).get(0).toString();
	executeSql("purge binary logs to '" + binlogName + "'");
	// restore task
	CompletableFuture.supplyAsync(
	() -> {
	try {
	container.restoreJob(
	"/mysqlcdc_to_mysql_with_binlog_delete.conf",
	String.valueOf(jobId));
	} catch (Exception e) {
	log.error("Commit task exception :" + e.getMessage());
	throw new RuntimeException(e);
	}
	return null;
	});
	await().atMost(2, TimeUnit.MINUTES)
	.untilAsserted(
	() -> {
	String jobStatus = container.getJobStatus(String.valueOf(jobId));
	Assertions.assertEquals("RUNNING", jobStatus);
	});
	// write data again, check no problem
	upsertDeleteSourceTable(MYSQL_DATABASE, SOURCE_TABLE);
	await().atMost(60000, TimeUnit.MILLISECONDS)
	.untilAsserted(
	() -> {
	Assertions.assertIterableEquals(
	query(getSourceQuerySQL(MYSQL_DATABASE, SOURCE_TABLE)),
	query(getSinkQuerySQL(MYSQL_DATABASE, SINK_TABLE)));
	});
	// check job status is not failed
	await().pollDelay(20, TimeUnit.SECONDS)
	.atMost(60, TimeUnit.SECONDS)
	.untilAsserted(
	() ->
	Assertions.assertEquals(
	"RUNNING", container.getJobStatus(String.valueOf(jobId))));
	// cancel task
	Assertions.assertEquals(0, container.cancelJob(String.valueOf(jobId)).getExitCode());
	}

I believe this test already covers savepoint and restore for CDC jobs to some extent. If you have any specific scenarios or additional cases in mind, could you please let me know?