[Bug][Zeta] Unchecked lifecycle close failure skips remaining task resource cleanup

unchecked=false, attempt=1, laterClosed=true, executorShutdown=true
unchecked=false, attempt=2, laterClosed=true, executorShutdown=true
unchecked=false, attempt=3, laterClosed=true, executorShutdown=true
unchecked=true, attempt=1, laterClosed=false, executorShutdown=false
unchecked=true, attempt=2, laterClosed=false, executorShutdown=false
unchecked=true, attempt=3, laterClosed=false, executorShutdown=false

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import org.apache.seatunnel.api.common.metrics.MetricsContext;
import org.apache.seatunnel.engine.common.utils.concurrent.CompletableFuture;
import org.apache.seatunnel.engine.core.dag.actions.SourceAction;
import org.apache.seatunnel.engine.server.dag.physical.config.SourceConfig;
import org.apache.seatunnel.engine.server.execution.ProgressState;
import org.apache.seatunnel.engine.server.task.SeaTunnelTask;
import org.apache.seatunnel.engine.server.task.flow.FlowLifeCycle;
import org.apache.seatunnel.engine.server.task.flow.SourceFlowLifeCycle;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;

/**
 * Exercises the unchanged production teardown loop with a failing lifecycle before an owned
 * executor. Every worker created by the harness is explicitly stopped in finally.
 */
public final class TaskCloseLeakRepro {
    public static void main(String[] args) throws Exception {
        run(false);
        run(true);
    }

    private static void run(boolean unchecked) throws Exception {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        CountDownLatch started = new CountDownLatch(1);
        CountDownLatch stop = new CountDownLatch(1);
        AtomicBoolean laterClosed = new AtomicBoolean();
        executor.submit(() -> {
            started.countDown();
            try {
                stop.await();
            } catch (InterruptedException expected) {
                Thread.currentThread().interrupt();
            }
        });
        if (!started.await(5, TimeUnit.SECONDS)) throw new AssertionError("Worker did not start");
        try {
            FlowLifeCycle first = new FlowLifeCycle() {
                @Override
                public void close() throws IOException {
                    if (unchecked) throw new IllegalStateException("injected sink close failure");
                    throw new IOException("injected checked close failure");
                }
            };
            FlowLifeCycle second = new FlowLifeCycle() {
                @Override
                public void close() {
                    laterClosed.set(true);
                    executor.shutdownNow();
                }
            };
            Task task = new Task(first, second);
            for (int attempt = 1; attempt <= 3; attempt++) {
                try {
                    task.close();
                    throw new AssertionError("Original close failure was swallowed");
                } catch (IOException | IllegalStateException expected) {
                    Logger.getLogger("repro").info(
                            "unchecked=" + unchecked + ", attempt=" + attempt
                                    + ", laterClosed=" + laterClosed.get()
                                    + ", executorShutdown=" + executor.isShutdown());
                }
                if (laterClosed.get() == unchecked || executor.isShutdown() == unchecked) {
                    throw new AssertionError("Unexpected teardown behavior");
                }
            }
        } finally {
            stop.countDown();
            executor.shutdownNow();
            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                throw new AssertionError("Harness cleanup failed");
            }
        }
    }

    /**
     * Supplies only the lifecycle list; task.close executes the real SeaTunnelTask method and
     * AbstractTask cleanup, with no Hazelcast member or connector process.
     */
    private static final class Task extends SeaTunnelTask {
        private Task(FlowLifeCycle... cycles) {
            super(1L, null, 0, null, Collections.emptyMap());
            allCycles = Arrays.asList(cycles);
            restoreComplete = new CompletableFuture<>();
        }

        @Override
        protected SourceFlowLifeCycle<?, ?> createSourceFlowLifeCycle(
                SourceAction<?, ?, ?> action, SourceConfig config,
                CompletableFuture<Void> future, MetricsContext metrics) {
            throw new UnsupportedOperationException("No source initialization in this teardown test");
        }

        @Override
        public ProgressState call() {
            throw new UnsupportedOperationException("No task scheduling in this teardown test");
        }

        @Override
        protected void collect() {
            throw new UnsupportedOperationException("No record processing in this teardown test");
        }
    }
}