[Fix][Zeta] Fix local mode can not finish

[Hisoka-X]

Purpose of this pull request

Fix local mode can not finish cause by #9532 .

Does this PR introduce any user-facing change?

no

How was this patch tested?

add new test

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

[hawk9821]

LGTM

[Copilot]

Pull Request Overview

This PR addresses an issue where local mode does not finish by improving interruption handling in the pending job scheduler thread and adding a test to verify thread lifecycle.

• Add specific InterruptedException catch in CoordinatorService to avoid swallowing interrupts.
• Enhance CoordinatorServiceTest to assert that the scheduler thread starts and stops correctly.
• Refactor test assertions for clarity.

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
CoordinatorService.java	Handle InterruptedException explicitly in the scheduler thread loop.
CoordinatorServiceTest.java	Capture initial thread count, assert it’s >0, then verify count decrements after shutdown; changed assertions to assertEquals.

Comments suppressed due to low confidence (1)

seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java:202

• [nitpick] Rename scheduleRunnerThreadCount to initialSchedulerThreadCount for clearer intent in the setup phase of the test.

        int scheduleRunnerThreadCount =

[Copilot]

Instead of throwing a RuntimeException on interrupt, consider restoring the thread's interrupted status with Thread.currentThread().interrupt() and breaking the loop to allow graceful shutdown.

Suggested change

	throw new RuntimeException(interrupted);
	Thread.currentThread().interrupt();
	break;

[Copilot]

[nitpick] This direct count check may be flaky. Use an awaitility or join on the scheduler thread with a timeout to ensure it has stopped before asserting the thread count.

Suggested change

	Assertions.assertEquals(
	scheduleRunnerThreadCount - 1,
	Thread.getAllStackTraces().keySet().stream()
	.filter(
	thread ->
	thread.getName().startsWith("pending-job-schedule-runner"))
	.count());
	await().atMost(10, TimeUnit.SECONDS)
	.untilAsserted(() -> Assertions.assertEquals(
	scheduleRunnerThreadCount - 1,
	Thread.getAllStackTraces().keySet().stream()
	.filter(
	thread ->
	thread.getName().startsWith("pending-job-schedule-runner"))
	.count()));

[davidzollo]

+1