[Feature-16300][registry] Added Consul-based Register implementation

dolphinscheduler-bom/pom.xml

@@ -124,6 +124,7 @@
         <zeppelin-client.version>0.10.1</zeppelin-client.version>
         <aliyun-voice.version>2.1.4</aliyun-voice.version>
         <nashorn-sandbox.version>0.3.2</nashorn-sandbox.version>
+        <consul-client.version>1.5.3</consul-client.version>
     </properties>
 
     <dependencyManagement>
@@ -982,6 +983,12 @@
                 <artifactId>delight-nashorn-sandbox</artifactId>
                 <version>${nashorn-sandbox.version}</version>
             </dependency>
+
+            <dependency>
+                <groupId>com.orbitz.consul</groupId>
+                <artifactId>consul-client</artifactId>
+                <version>${consul-client.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

dolphinscheduler-registry/dolphinscheduler-registry-all/pom.xml

@@ -42,5 +42,10 @@
             <artifactId>dolphinscheduler-registry-etcd</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-registry-consul</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 </project>

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-consul/README.md

@@ -0,0 +1,22 @@
+# Introduction
+
+This module is the consul registry plugin module, this plugin will use consul as the registry center.
+
+# How to use
+
+If you want to set the registry center as consul,you need to set the registry properties in master/worker/api's application.yml
+
+```yaml
+registry:
+  type: consul
+  url: http://localhost:8500
+  namespace: dolphinscheduler
+  session-timeout: 30s
+  session-refresh-time: 3s
+  user-name: 
+  password:
+  acl-token:
+```
+
+After do this config, you can start your DolphinScheduler cluster, your cluster will use consul as registry center to
+store server metadata.

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-consul/pom.xml

@@ -0,0 +1,53 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.dolphinscheduler</groupId>
+        <artifactId>dolphinscheduler-registry-plugins</artifactId>
+        <version>dev-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>dolphinscheduler-registry-consul</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-registry-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-common</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-registry-it</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.orbitz.consul</groupId>
+            <artifactId>consul-client</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+        </dependency>
+
+    </dependencies>
+</project>

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-consul/src/main/java/org/apache/dolphinscheduler/plugin/registry/consul/ConsulConnectionStateListener.java

@@ -0,0 +1,112 @@
+/*
+ * 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.dolphinscheduler.plugin.registry.consul;
+
+import org.apache.dolphinscheduler.registry.api.ConnectionListener;
+import org.apache.dolphinscheduler.registry.api.ConnectionState;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.orbitz.consul.Consul;
+
+public class ConsulConnectionStateListener implements AutoCloseable {
+
+    private final List<ConnectionListener> connectionListeners = Collections.synchronizedList(new ArrayList<>());
+    // A thread pool that periodically obtains connection status
+    private final ScheduledExecutorService scheduledExecutorService;
+    // monitored client
+    private final Consul consulClient;
+    // The state of the last monitor
+    private volatile ConnectionState connectionState;
+
+    public ConsulConnectionStateListener(Consul consulClient) {
+        this.consulClient = consulClient;
+        this.scheduledExecutorService = Executors.newScheduledThreadPool(
+                1,
+                new ThreadFactoryBuilder().setNameFormat("ConsulConnectionStateListenerThread")
+                        .setDaemon(true).build());
+    }
+
+    public void addConnectionListener(ConnectionListener connectionListener) {
+        connectionListeners.add(connectionListener);
+    }
+
+    @Override
+    public void close() {
+        connectionListeners.clear();
+        scheduledExecutorService.shutdownNow();
+    }
+
+    /**
+     * Apply for a lease through the client, if there is no exception, the connection is normal
+     * @return the current connection state
+     * @throws if there is an exception, return is DISCONNECTED
+     */
+    private ConnectionState currentConnectivityState() {
+        try {
+            consulClient.agentClient().ping();
+            return ConnectionState.CONNECTED;
+        } catch (Exception e) {
+            return ConnectionState.DISCONNECTED;
+        }
+    }
+
+    /**
+     * Periodically execute thread to get connection status
+     */
+    public void start() {
+        long initialDelay = 500L;
+        long delay = 500L;
+        this.scheduledExecutorService.scheduleWithFixedDelay(() -> {
+            ConnectionState currentConnectionState = currentConnectivityState();
+            if (currentConnectionState == connectionState) {
+                return;
+            }
+            if (connectionState == ConnectionState.CONNECTED) {
+                if (currentConnectionState == ConnectionState.DISCONNECTED) {
+                    connectionState = ConnectionState.DISCONNECTED;
+                    triggerListener(ConnectionState.DISCONNECTED);
+                }
+            } else if (connectionState == ConnectionState.DISCONNECTED) {
+                if (currentConnectionState == ConnectionState.CONNECTED) {
+                    connectionState = ConnectionState.CONNECTED;
+                    triggerListener(ConnectionState.RECONNECTED);
+                }
+            } else if (connectionState == null) {
+                connectionState = currentConnectionState;
+                triggerListener(connectionState);
+            }
+        },
+                initialDelay,
+                delay,
+                TimeUnit.MILLISECONDS);
+    }
+
+    // notify all listeners
+    private void triggerListener(ConnectionState connectionState) {
+        for (ConnectionListener connectionListener : connectionListeners) {
+            connectionListener.onUpdate(connectionState);
+        }
+    }
+}