[Chore] Bump testcontainer to 1.21.4 to fix could not find a valid Docker environment at CI (#17978)

Author: ruanwenjun

dolphinscheduler-api-test/pom.xml

@@ -37,7 +37,7 @@
 
         <junit.version>5.7.2</junit.version>
         <selenium.version>4.21.0</selenium.version>
-        <testcontainers.version>1.19.8</testcontainers.version>
+        <testcontainers.version>1.21.4</testcontainers.version>
         <lombok.version>1.18.24</lombok.version>
         <assertj-core.version>3.23.1</assertj-core.version>
         <awaitility.version>4.1.0</awaitility.version>

dolphinscheduler-bom/pom.xml

@@ -119,7 +119,7 @@
         <qcloud-cos.version>5.6.231</qcloud-cos.version>
         <system-lambda.version>1.2.1</system-lambda.version>
         <zeppelin-client.version>0.10.1</zeppelin-client.version>
-        <testcontainer.version>1.19.3</testcontainer.version>
+        <testcontainer.version>1.21.4</testcontainer.version>
         <checker-qual.version>3.19.0</checker-qual.version>
         <zeppelin-client.version>0.10.1</zeppelin-client.version>
         <aliyun-voice.version>2.1.4</aliyun-voice.version>

dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/DateUtilsTest.java

@@ -30,18 +30,15 @@
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 public class DateUtilsTest {
 
-    @BeforeEach
-    public void before() {
-        ThreadLocalContext.removeTimezone();
-    }
+    private final TimeZone defaultTimeZone = TimeZone.getDefault();
 
     @AfterEach
-    public void after() {
+    public void rollbackTimeZone() {
+        TimeZone.setDefault(defaultTimeZone);
         ThreadLocalContext.removeTimezone();
     }
 

dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/JSONUtilsTest.java

@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.TimeZone;
 
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -41,6 +42,14 @@
 
 public class JSONUtilsTest {
 
+    private final TimeZone defaultTimeZone = TimeZone.getDefault();
+
+    @AfterEach
+    public void rollbackTimeZone() {
+        TimeZone.setDefault(defaultTimeZone);
+        JSONUtils.setTimeZone(defaultTimeZone);
+    }
+
     @Test
     public void createObjectNodeTest() {
         String jsonStr = "{\"a\":\"b\",\"b\":\"d\"}";

dolphinscheduler-e2e/pom.xml

@@ -44,7 +44,7 @@
         <log4j-slf4j-impl.version>2.17.2</log4j-slf4j-impl.version>
         <guava.version>31.0.1-jre</guava.version>
         <maven-surefire-plugin.version>2.22.2</maven-surefire-plugin.version>
-        <testcontainers.version>1.19.8</testcontainers.version>
+        <testcontainers.version>1.21.4</testcontainers.version>
         <junit-pioneer.version>2.2.0</junit-pioneer.version>
     </properties>
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/LogicTaskExecutorContainerProvider.java

@@ -18,17 +18,18 @@
 package org.apache.dolphinscheduler.server.master.engine.executor;
 
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
-import org.apache.dolphinscheduler.task.executor.container.ITaskExecutorContainer;
 import org.apache.dolphinscheduler.task.executor.container.ITaskExecutorContainerProvider;
 import org.apache.dolphinscheduler.task.executor.container.SharedThreadTaskExecutorContainer;
 import org.apache.dolphinscheduler.task.executor.container.TaskExecutorContainerConfig;
 
 import org.springframework.stereotype.Component;
 
 @Component
-public class LogicTaskExecutorContainerProvider implements ITaskExecutorContainerProvider {
+public class LogicTaskExecutorContainerProvider
+        implements
+            ITaskExecutorContainerProvider<SharedThreadTaskExecutorContainer> {
 
-    private final ITaskExecutorContainer taskExecutorContainer;
+    private final SharedThreadTaskExecutorContainer taskExecutorContainer;
 
     public LogicTaskExecutorContainerProvider(final MasterConfig masterConfig) {
         final TaskExecutorContainerConfig containerConfig = TaskExecutorContainerConfig.builder()
@@ -39,7 +40,7 @@ public LogicTaskExecutorContainerProvider(final MasterConfig masterConfig) {
     }
 
     @Override
-    public ITaskExecutorContainer getExecutorContainer() {
+    public SharedThreadTaskExecutorContainer getExecutorContainer() {
         return taskExecutorContainer;
     }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java

@@ -34,6 +34,8 @@
 
 import lombok.extern.slf4j.Slf4j;
 
+import com.google.common.annotations.VisibleForTesting;
+
 /**
  * WorkerGroupTaskDispatcher is responsible for dispatching tasks from the task queue.
  * The main responsibilities include:
@@ -179,7 +181,13 @@ public synchronized void close() {
         }
     }
 
-    int queueSize() {
+    @VisibleForTesting
+    public int dispatchEventCount() {
         return this.workerGroupEventBus.size();
     }
+
+    @VisibleForTesting
+    public int waitingDispatchTaskCount() {
+        return this.waitingDispatchTaskIds.size();
+    }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinator.java

@@ -21,13 +21,16 @@
 import org.apache.dolphinscheduler.server.master.engine.task.client.ITaskExecutorClient;
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 import lombok.extern.slf4j.Slf4j;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import com.google.common.annotations.VisibleForTesting;
+
 /**
  * WorkerGroupTaskDispatcherManager is responsible for managing the task dispatching for worker groups.
  * It maintains a mapping of worker groups to their task dispatchers and priority delay queues,
@@ -40,12 +43,12 @@ public class WorkerGroupDispatcherCoordinator implements AutoCloseable {
     @Autowired
     private ITaskExecutorClient taskExecutorClient;
 
-    private final ConcurrentHashMap<String, WorkerGroupDispatcher> workerGroupDispatcherMap;
+    private final Map<String, WorkerGroupDispatcher> workerGroupDispatchers;
 
     private final MasterConfig masterConfig;
 
     public WorkerGroupDispatcherCoordinator(final MasterConfig masterConfig) {
-        workerGroupDispatcherMap = new ConcurrentHashMap<>();
+        workerGroupDispatchers = new ConcurrentHashMap<>();
         this.masterConfig = masterConfig;
     }
 
@@ -82,7 +85,12 @@ public boolean removeTask(ITaskExecutionRunnable taskExecutionRunnable) {
     }
 
     public boolean existWorkerGroup(String workerGroup) {
-        return workerGroupDispatcherMap.containsKey(workerGroup);
+        return workerGroupDispatchers.containsKey(workerGroup);
+    }
+
+    @VisibleForTesting
+    public Map<String, WorkerGroupDispatcher> workerGroupDispatchers() {
+        return workerGroupDispatchers;
     }
 
     /**
@@ -91,7 +99,7 @@ public boolean existWorkerGroup(String workerGroup) {
     @Override
     public void close() throws Exception {
         log.info("WorkerGroupDispatcherCoordinator closing");
-        for (WorkerGroupDispatcher workerGroupDispatcher : workerGroupDispatcherMap.values()) {
+        for (WorkerGroupDispatcher workerGroupDispatcher : workerGroupDispatchers.values()) {
             try {
                 workerGroupDispatcher.close();
             } catch (Exception e) {
@@ -102,7 +110,7 @@ public void close() throws Exception {
     }
 
     private WorkerGroupDispatcher getOrCreateWorkerGroupDispatcher(String workerGroup) {
-        return workerGroupDispatcherMap.computeIfAbsent(workerGroup, wg -> {
+        return workerGroupDispatchers.computeIfAbsent(workerGroup, wg -> {
             WorkerGroupDispatcher workerGroupDispatcher =
                     new WorkerGroupDispatcher(wg, taskExecutorClient, masterConfig.getTaskDispatchPolicy());
             workerGroupDispatcher.start();

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherTest.java

@@ -34,7 +34,6 @@
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
-import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.config.TaskDispatchPolicy;
 import org.apache.dolphinscheduler.server.master.engine.WorkflowEventBus;
 import org.apache.dolphinscheduler.server.master.engine.task.client.ITaskExecutorClient;
@@ -49,6 +48,7 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.InOrder;
@@ -61,9 +61,14 @@ class WorkerGroupDispatcherTest {
     @BeforeEach
     void setUp() {
         taskExecutorClient = mock(ITaskExecutorClient.class);
-        final MasterConfig masterConfig = new MasterConfig();
         dispatcher =
-                new WorkerGroupDispatcher("TestGroup", taskExecutorClient, masterConfig.getTaskDispatchPolicy());
+                new WorkerGroupDispatcher("TestGroup", taskExecutorClient, new TaskDispatchPolicy());
+    }
+
+    @AfterEach
+    void tearDown() {
+        dispatcher.close();
+        dispatcher.interrupt();
     }
 
     @Test

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/MasterContainer.java

@@ -27,9 +27,9 @@
 import org.apache.dolphinscheduler.server.master.engine.executor.LogicTaskExecutorLifecycleEventReporter;
 import org.apache.dolphinscheduler.server.master.engine.executor.LogicTaskExecutorRepository;
 import org.apache.dolphinscheduler.server.master.engine.system.SystemEventBus;
-import org.apache.dolphinscheduler.task.executor.container.AbstractTaskExecutorContainer;
+import org.apache.dolphinscheduler.server.master.engine.task.dispatcher.WorkerGroupDispatcherCoordinator;
+import org.apache.dolphinscheduler.task.executor.container.SharedThreadTaskExecutorContainer;
 import org.apache.dolphinscheduler.task.executor.container.TaskExecutorAssignmentTable;
-import org.apache.dolphinscheduler.task.executor.worker.TaskExecutorWorkers;
 
 import java.util.concurrent.TimeUnit;
 
@@ -57,31 +57,59 @@ public class MasterContainer {
     @Autowired
     private LogicTaskExecutorLifecycleEventReporter logicTaskExecutorLifecycleEventReporter;
 
+    @Autowired
+    private WorkerGroupDispatcherCoordinator workerGroupDispatcherCoordinator;
+
     public void assertAllResourceReleased() {
         await()
                 .atMost(10, TimeUnit.SECONDS)
                 .untilAsserted(this::doAssertAllResourceReleased);
     }
 
     private void doAssertAllResourceReleased() {
+        assertWorkflowReleased();
+        assertWorkflowEventBusReleased();
+
+        assertSystemEventBusReleased();
+
+        assertLogicTaskEngineReleased();
+
+        assertWorkerGroupDispatcherReleased();
+    }
+
+    private void assertWorkflowReleased() {
         assertThat(workflowRepository.getAll()).isEmpty();
+        assertThat(workflowEventBusFireWorkers.getWorkers())
+                .allMatch(worker -> worker.getRegisteredWorkflowExecuteRunnableSize() == 0);
+    }
 
+    private void assertWorkflowEventBusReleased() {
         assertThat(workflowEventBusFireWorkers.getWorkers())
-                .allMatch(workflowEventBusFireWorker -> workflowEventBusFireWorker
-                        .getRegisteredWorkflowExecuteRunnableSize() == 0);
+                .allMatch(worker -> worker.getRegisteredWorkflowExecuteRunnableSize() == 0);
+    }
+
+    private void assertSystemEventBusReleased() {
         assertThat(systemEventBus).matches(AbstractDelayEventBus::isEmpty);
+    }
 
+    private void assertLogicTaskEngineReleased() {
         assertThat(logicTaskExecutorRepository.getAll()).isEmpty();
 
-        final AbstractTaskExecutorContainer executorContainer =
-                (AbstractTaskExecutorContainer) logicTaskExecutorContainerProvider.getExecutorContainer();
+        final SharedThreadTaskExecutorContainer executorContainer =
+                logicTaskExecutorContainerProvider.getExecutorContainer();
         assertThat(executorContainer.getTaskExecutorAssignmentTable()).matches(TaskExecutorAssignmentTable::isEmpty);
 
-        final TaskExecutorWorkers taskExecutorWorkers = executorContainer.getTaskExecutorWorkers();
-        assertThat(taskExecutorWorkers.getWorkers())
+        assertThat(executorContainer.getTaskExecutorWorkers().getWorkers())
                 .allMatch(taskExecutorWorker -> taskExecutorWorker.getRegisteredTaskExecutorSize() == 0)
                 .allMatch(taskExecutorWorker -> taskExecutorWorker.getFiredTaskExecutorSize() == 0);
 
         assertThat(logicTaskExecutorLifecycleEventReporter.getEventChannels()).isEmpty();
     }
+
+    private void assertWorkerGroupDispatcherReleased() {
+        assertThat(workerGroupDispatcherCoordinator.workerGroupDispatchers().values())
+                .allMatch(dispatcher -> dispatcher.dispatchEventCount() == 0)
+                .allMatch(dispatcher -> dispatcher.waitingDispatchTaskCount() == 0);
+    }
+
 }