wait all process stop

Author: skaic

elasticjob-ecosystem/elasticjob-executor/elasticjob-executor-kernel/src/main/java/org/apache/shardingsphere/elasticjob/executor/ElasticJobExecutor.java

@@ -38,6 +38,7 @@
 import java.util.Queue;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
@@ -141,14 +142,19 @@ private void execute(final JobConfiguration jobConfig, final ShardingContexts sh
     private void process(final JobConfiguration jobConfig, final ShardingContexts shardingContexts, final ExecutionSource executionSource) {
         Collection<Integer> items = shardingContexts.getShardingItemParameters().keySet();
         Queue<Future<Boolean>> futures = new LinkedList<>();
+        CountDownLatch latch = new CountDownLatch(items.size());
         for (int each : items) {
             JobExecutionEvent jobExecutionEvent = new JobExecutionEvent(IpUtils.getHostName(), IpUtils.getIp(), shardingContexts.getTaskId(), jobConfig.getJobName(), executionSource, each);
             ExecutorService executorService = executorContext.get(ExecutorService.class);
             if (executorService.isShutdown()) {
                 return;
             }
             Future<Boolean> future = executorService.submit(() -> {
-                process(jobConfig, shardingContexts, each, jobExecutionEvent);
+                try {
+                    process(jobConfig, shardingContexts, each, jobExecutionEvent);
+                } finally {
+                    latch.countDown();
+                }
                 return true;
             });
             futures.offer(future);
@@ -179,7 +185,12 @@ private void process(final JobConfiguration jobConfig, final ShardingContexts sh
             } finally {
                 futures.poll();
             }
+        }
 
+        try {
+            latch.await();
+        } catch (final InterruptedException ex) {
+            Thread.currentThread().interrupt();
         }
     }