Fix fragment failure propagation race

[JackieTien97]

Summary

• serialize ShuffleSinkHandle termination with an explicit owner and completion signal
• make concurrent abort callers wait until channel cleanup is complete before fragment memory can be deregistered
• suppress normal end-of-block notifications while abort or close is in progress
• preserve non-blocking reentrant close behavior to avoid channel-lock deadlocks

Root cause

A failed fragment can be cleaned concurrently by DriverScheduler.forceDeregisterFragmentInstance and the asynchronous FragmentInstanceExecution state listener. The first abort owns channel cleanup, while the second abort previously returned false immediately. The state listener could then close its drivers and deregister fragment memory before the first abort finished. Driver.close could call setNoMoreTsBlocks while aborted was still false, sending a normal end-of-stream to the coordinator and occasionally hiding the execution error from the client. Concurrent memory deregistration also explains the observed MemoryLeakException and missing RelatedMemoryReserved entries.

This change makes the abort completion itself the synchronization boundary. A competing abort waits for the owner to complete, while close remains non-blocking because it may be re-entered from a channel callback that already holds a channel lock. Normal end-of-block notifications are rejected whenever termination has been claimed.

Reproduced from the flaky CI failure: https://github.com/apache/iotdb/actions/runs/33285987416/job/99189163860

Tests

• mvn test -pl iotdb-core/datanode -Dtest=ShuffleSinkHandleTest -DfailIfNoTests=false with JDK 17
• IoTDBTableAggregationNonStreamIT#rateFunctionsExceptionTest under TableClusterIT with JDK 17, four consecutive successful 1C3D runs
• verified the four DataNode log sets contain no MemoryLeakException or missing RelatedMemoryReserved errors