/*
 * 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.
 */

package org.apache.shardingsphere.test.integration.engine.junit.parallel.impl;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.test.integration.engine.junit.parallel.ParallelRunnerExecutor;
import org.apache.shardingsphere.test.integration.engine.param.model.ParameterizedArray;

import java.io.Closeable;
import java.util.Collection;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Parallel runner executor with scenario.
 */
@Slf4j
public final class ScenarioParallelRunnerExecutor implements ParallelRunnerExecutor {
    
    private final Map<ScenarioKey, ScenarioExecutorService> executorServices = new ConcurrentHashMap<>();
    
    private final Collection<Future<?>> taskFeatures = new LinkedList<>();
    
    @Override
    public void execute(final ParameterizedArray parameterizedArray, final Runnable childStatement) {
        taskFeatures.add(getExecutorService(new ScenarioKey(parameterizedArray)).submit(childStatement));
    }
    
    private ScenarioExecutorService getExecutorService(final ScenarioKey scenarioKey) {
        if (executorServices.containsKey(scenarioKey)) {
            return executorServices.get(scenarioKey);
        }
        ScenarioExecutorService newExecutorService = new ScenarioExecutorService(scenarioKey);
        if (null != executorServices.putIfAbsent(scenarioKey, newExecutorService)) {
            newExecutorService.close();
        }
        return executorServices.get(scenarioKey);
    }
    
    @Override
    public void finished() {
        taskFeatures.forEach(future -> {
            try {
                future.get();
            } catch (final InterruptedException | ExecutionException ignored) {
            }
        });
        executorServices.values().forEach(ScenarioExecutorService::close);
    }
    
    /**
     * Scenario key.
     */
    @EqualsAndHashCode
    private static final class ScenarioKey {
        
        private final String adapter;
        
        private final String scenario;
        
        private final String databaseTypeName;
        
        private ScenarioKey(final ParameterizedArray parameterizedArray) {
            adapter = parameterizedArray.getAdapter();
            scenario = parameterizedArray.getScenario();
            databaseTypeName = parameterizedArray.getDatabaseType().getName();
        }
        
        @Override
        public String toString() {
            return String.join("-", adapter, scenario, databaseTypeName);
        }
    }
    
    /**
     * Scenario executor service.
     */
    private static final class ScenarioExecutorService implements Closeable {
        
        private final ExecutorService executorService;
        
        ScenarioExecutorService(final ScenarioKey scenarioKey) {
            String threadPoolNameFormat = String.join("-", "ScenarioExecutorPool", scenarioKey.toString(), "%d");
            executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new ThreadFactoryBuilder().setNameFormat(threadPoolNameFormat).build());
        }
        
        /**
         * Submit task.
         * 
         * @param childStatement child statement
         * @return task future
         */
        public Future<?> submit(final Runnable childStatement) {
            return executorService.submit(childStatement);
        }
        
        @Override
        public void close() {
            executorService.shutdownNow();
        }
    }
}

@@ -887,129 +887,118 @@
 ort 
-com.google.common.util.concurrent.ThreadFactoryBuilder;%0Aimport lombok.EqualsAndHashCode;%0Aimport lombok.extern.slf4j.Slf4j
+lombok.EqualsAndHashCode;%0Aimport org.apache.shardingsphere.infra.executor.kernel.thread.ExecutorServiceManager
 ;%0Aim
@@ -1185,34 +1185,8 @@
 y;%0A%0A
-import java.io.Closeable;%0A
 impo
@@ -1388,194 +1388,14 @@
 ent.
-ExecutorService;%0Aimport java.util.concurrent.Future;%0Aimport java.util.concurrent.LinkedBlockingQueue;%0Aimport java.util.concurrent.ThreadPoolExecutor;%0Aimport java.util.concurrent.TimeUnit
+Future
 ;%0A%0A/
@@ -1448,15 +1448,8 @@
  */%0A
-@Slf4j%0A
 publ
@@ -1570,24 +1570,16 @@
 rioKey, 
-Scenario
 Executor
@@ -1585,16 +1585,23 @@
 rService
+Manager
 %3E execut
@@ -1609,16 +1609,23 @@
 rService
+Manager
 s = new 
@@ -1929,16 +1929,37 @@
 Array)).
+getExecutorService().
 submit(c
@@ -1998,24 +1998,16 @@
 private 
-Scenario
 Executor
@@ -2013,16 +2013,23 @@
 rService
+Manager
  getExec
@@ -2100,16 +2100,23 @@
 rService
+Manager
 s.contai
@@ -2163,32 +2163,39 @@
  executorService
+Manager
 s.get(scenarioKe
@@ -2221,91 +2221,218 @@
    S
-cenarioExecutorService newExecutorService = new ScenarioExecutorService(scenarioKey
+tring threadPoolNameFormat = String.join(%22-%22, %22ScenarioExecutorPool%22, scenarioKey.toString(), %22%25d%22);%0A        ExecutorServiceManager newExecutorServiceManager = new ExecutorServiceManager(1, threadPoolNameFormat
 );%0A 
@@ -2465,16 +2465,23 @@
 rService
+Manager
 s.putIfA
@@ -2517,16 +2517,23 @@
 rService
+Manager
 )) %7B%0A   
@@ -2559,16 +2559,23 @@
 rService
+Manager
 .close()
@@ -2616,16 +2616,23 @@
 rService
+Manager
 s.get(sc
@@ -2726,22 +2726,20 @@
 forEach(
-future
+each
  -%3E %7B%0A  
@@ -2770,22 +2770,20 @@
         
-future
+each
 .get();%0A
@@ -2911,16 +2911,23 @@
 rService
+Manager
 s.values
@@ -2941,38 +2941,55 @@
 ach(
-ScenarioExecutorService::close
+each -%3E each.getExecutorService().shutdownNow()
 );%0A 
@@ -3695,988 +3695,6 @@
   %7D%0A
-    %0A    /**%0A     * Scenario executor service.%0A     */%0A    private static final class ScenarioExecutorService implements Closeable %7B%0A        %0A        private final ExecutorService executorService;%0A        %0A        ScenarioExecutorService(final ScenarioKey scenarioKey) %7B%0A            String threadPoolNameFormat = String.join(%22-%22, %22ScenarioExecutorPool%22, scenarioKey.toString(), %22%25d%22);%0A            executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue%3C%3E(), new ThreadFactoryBuilder().setNameFormat(threadPoolNameFormat).build());%0A        %7D%0A        %0A        /**%0A         * Submit task.%0A         * %0A         * @param childStatement child statement%0A         * @return task future%0A         */%0A        public Future%3C?%3E submit(final Runnable childStatement) %7B%0A            return executorService.submit(childStatement);%0A        %7D%0A        %0A        @Override%0A        public void close() %7B%0A            executorService.shutdownNow();%0A        %7D%0A    %7D%0A
 %7D%0A

Remove ScenarioParallelRunnerExecutor.ScenarioExecutorService (#9489)