This is an automated email from the ASF dual-hosted git repository.
terrymanu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere-elasticjob.git
The following commit(s) were added to refs/heads/master by this push:
new 4ad7d52fd Add Etcd as the registry center (#2517)
4ad7d52fd is described below
commit 4ad7d52fd5ef93e459f781a6f3127aec97816e3d
Author: Hengqian Ling <[email protected]>
AuthorDate: Sun Jun 28 11:37:39 2026 +0800
Add Etcd as the registry center (#2517)
---
RELEASE-NOTES.md | 1 +
.../content/user-manual/configuration/_index.cn.md | 32 +-
.../content/user-manual/configuration/_index.en.md | 32 +-
.../configuration/registry-center/_index.cn.md | 75 ++++
.../configuration/registry-center/_index.en.md | 75 ++++
.../configuration/registry-center/etcd.cn.md | 98 +++++
.../configuration/registry-center/etcd.en.md | 98 +++++
.../configuration/registry-center/zookeeper.cn.md | 118 +++++
.../configuration/registry-center/zookeeper.en.md | 118 +++++
lifecycle/pom.xml | 6 +
.../internal/reg/RegistryCenterFactory.java | 15 +-
pom.xml | 7 +
.../elasticjob/reg/spi/RegistryCenterCreator.java | 46 ++
.../provider/etcd}/pom.xml | 31 +-
.../elasticjob/reg/etcd/EtcdConfiguration.java | 73 ++++
.../elasticjob/reg/etcd/EtcdRegistryCenter.java | 481 +++++++++++++++++++++
.../reg/etcd/EtcdRegistryCenterCreator.java | 46 ++
.../exception/EtcdIgnoredExceptionProvider.java | 36 ++
...asticjob.reg.exception.IgnoredExceptionProvider | 18 +
...sphere.elasticjob.reg.spi.RegistryCenterCreator | 18 +
.../elasticjob/reg/etcd/EtcdConfigurationTest.java | 63 +++
.../reg/etcd/EtcdRegistryCenterCreatorTest.java | 49 +++
.../EtcdIgnoredExceptionProviderTest.java | 40 ++
registry-center/provider/pom.xml | 1 +
.../zookeeper/ZookeeperRegistryCenterCreator.java | 42 ++
...sphere.elasticjob.reg.spi.RegistryCenterCreator | 18 +
26 files changed, 1592 insertions(+), 45 deletions(-)
diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md
index 955a17e3a..68df985c5 100644
--- a/RELEASE-NOTES.md
+++ b/RELEASE-NOTES.md
@@ -3,6 +3,7 @@
### Enhancements
1. Bump the JDK requirement for build time from JDK 8 to JDK 17 -
[#2509](https://github.com/apache/shardingsphere-elasticjob/issues/2509)
+1. Registry Center: Support etcd3 as a registry center -
[#2221](https://github.com/apache/shardingsphere-elasticjob/issues/2221)
## 3.0.5
diff --git a/docs/content/user-manual/configuration/_index.cn.md
b/docs/content/user-manual/configuration/_index.cn.md
index 19c1225cc..ea5517611 100644
--- a/docs/content/user-manual/configuration/_index.cn.md
+++ b/docs/content/user-manual/configuration/_index.cn.md
@@ -13,24 +13,30 @@ ElasticJob 提供了 3 种配置方式，用于不同的使用场景。
## 注册中心配置项
-### 可配置属性
+ElasticJob
支持多种注册中心类型，详细配置请参见[注册中心配置](/cn/user-manual/configuration/registry-center)。
+
+### 支持的注册中心类型
+
+| 注册中心类型 | 说明 |
+|-------------|------|
+| [ZooKeeper](/cn/user-manual/configuration/registry-center/zookeeper) |
Apache ZooKeeper，分布式协调服务 |
+| [etcd](/cn/user-manual/configuration/registry-center/etcd) | etcd3，分布式键值存储 |
+
+### 通用配置属性
-| 属性名 | 类型 | 缺省值 | 描述 |
-|-------------------------------|:-------|:------|:--------------------|
-| serverLists | String | | 连接 ZooKeeper 服务器的列表 |
-| namespace | String | | ZooKeeper 的命名空间 |
-| baseSleepTimeMilliseconds | int | 1000 | 等待重试的间隔时间的初始毫秒数 |
-| maxSleepTimeMilliseconds | String | 3000 | 等待重试的间隔时间的最大毫秒数 |
-| maxRetries | String | 3 | 最大重试次数 |
-| sessionTimeoutMilliseconds | int | 60000 | 会话超时毫秒数 |
-| connectionTimeoutMilliseconds | int | 15000 | 连接超时毫秒数 |
-| digest | String | 无需验证 | 连接 ZooKeeper 的权限令牌 |
+| 属性名 | 类型 | 描述 |
+|-------------------------------|:-------|:--------------------|
+| serverLists | String | 连接注册中心服务器的列表 |
+| namespace | String | 注册中心的命名空间 |
+| connectionTimeoutMilliseconds | int | 连接超时毫秒数 |
### 核心配置项说明
**serverLists:**
-包括 IP 地址和端口号，多个地址用逗号分隔，如: host1:2181,host2:2181
+包括 IP 地址和端口号，多个地址用逗号分隔。
+- ZooKeeper: `host1:2181,host2:2181`
+- etcd: `http://host1:2379,http://host2:2379`
## 作业配置项
@@ -83,7 +89,7 @@ ElasticJob 提供了 3 种配置方式，用于不同的使用场景。
**reconcileIntervalMinutes:**
-在分布式的场景下由于网络、时钟等原因，可能导致 ZooKeeper 的数据与真实运行的作业产生不一致，这种不一致通过正向的校验无法完全避免。
+在分布式的场景下由于网络、时钟等原因，可能导致注册中心的数据与真实运行的作业产生不一致，这种不一致通过正向的校验无法完全避免。
需要另外启动一个线程定时校验注册中心数据与真实作业状态的一致性，即维持 ElasticJob 的最终一致性。
配置为小于 1 的任意值表示不执行修复。
diff --git a/docs/content/user-manual/configuration/_index.en.md
b/docs/content/user-manual/configuration/_index.en.md
index 8ea1858e0..16f648894 100644
--- a/docs/content/user-manual/configuration/_index.en.md
+++ b/docs/content/user-manual/configuration/_index.en.md
@@ -13,24 +13,30 @@ ElasticJob has provided 3 kinds of configuration methods
for different situation
## Registry Center Configuration
-### Configuration
+ElasticJob supports multiple types of registry centers. For detailed
configuration, please refer to [Registry Center
Configuration](/en/user-manual/configuration/registry-center).
+
+### Supported Registry Center Types
+
+| Registry Center Type | Description |
+|---------------------|-------------|
+| [ZooKeeper](/en/user-manual/configuration/registry-center/zookeeper) |
Apache ZooKeeper, distributed coordination service |
+| [etcd](/en/user-manual/configuration/registry-center/etcd) | etcd3,
distributed key-value store |
+
+### Common Configuration Properties
-| Name | Data Type | Default Value | Description
|
-|-------------------------------|:----------|:--------------|:---------------------------------------------------------|
-| serverLists | String | | ZooKeeper server
IP list |
-| namespace | String | | ZooKeeper
namespace |
-| baseSleepTimeMilliseconds | int | 1000 | The initial
value of milliseconds for the retry interval |
-| maxSleepTimeMilliseconds | String | 3000 | The maximum
value of milliseconds for the retry interval |
-| maxRetries | String | 3 | Maximum number
of retries |
-| sessionTimeoutMilliseconds | int | 60000 | Session timeout
in milliseconds |
-| connectionTimeoutMilliseconds | int | 15000 | Connection
timeout in milliseconds |
-| digest | String | no need | Permission token
to connect to ZooKeeper |
+| Name | Data Type | Description
|
+|-------------------------------|:----------|:---------------------------------------------------------|
+| serverLists | String | Registry center server IP list
|
+| namespace | String | Registry center namespace
|
+| connectionTimeoutMilliseconds | int | Connection timeout in
milliseconds |
### Core Configuration Description
**serverLists:**
-Include IP and port, multiple addresses are separated by commas, such as:
`host1:2181,host2:2181`
+Include IP and port, multiple addresses are separated by commas.
+- ZooKeeper: `host1:2181,host2:2181`
+- etcd: `http://host1:2379,http://host2:2379`
## Job Configuration
@@ -82,7 +88,7 @@ If the time error exceeds the configured seconds, an
exception will be thrown wh
**reconcileIntervalMinutes:**
-In a distributed system, due to network, clock and other reasons, ZooKeeper
may be inconsistent with the actual running job. This inconsistency cannot be
completely avoided through positive verification.
+In a distributed system, due to network, clock and other reasons, the registry
center may be inconsistent with the actual running job. This inconsistency
cannot be completely avoided through positive verification.
It is necessary to start another thread to periodically calibrate the
consistency between the registry center and the job status, that is, to
maintain the final consistency of ElasticJob.
Less than `1` means no repair is performed.
diff --git
a/docs/content/user-manual/configuration/registry-center/_index.cn.md
b/docs/content/user-manual/configuration/registry-center/_index.cn.md
new file mode 100644
index 000000000..767a0f29b
--- /dev/null
+++ b/docs/content/user-manual/configuration/registry-center/_index.cn.md
@@ -0,0 +1,75 @@
++++
+pre = "<b>4.1.2.1 </b>"
+title = "注册中心配置"
+weight = 1
+chapter = true
++++
+
+ElasticJob 支持多种注册中心类型，用于协调分布式作业的调度和执行。
+
+本章节介绍如何配置不同类型的注册中心。
+
+## 支持的注册中心类型
+
+| 注册中心类型 | 说明 |
+|-------------|------|
+| [ZooKeeper](/cn/user-manual/configuration/registry-center/zookeeper) |
Apache ZooKeeper，分布式协调服务 |
+| [etcd](/cn/user-manual/configuration/registry-center/etcd) | etcd3，分布式键值存储 |
+
+## 配置方式
+
+ElasticJob 提供了以下配置注册中心的方式：
+
+### Java API
+
+```java
+import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperConfiguration;
+import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+
+public class ZookeeperExample {
+
+ public static void main(String[] args) {
+ // ZooKeeper
+ ZookeeperConfiguration zkConfig = new
ZookeeperConfiguration("host1:2181,host2:2181", "elasticjob");
+ CoordinatorRegistryCenter registryCenter = new
ZookeeperRegistryCenter(zkConfig);
+ registryCenter.init();
+ }
+}
+```
+
+```java
+import org.apache.shardingsphere.elasticjob.reg.etcd.EtcdConfiguration;
+import org.apache.shardingsphere.elasticjob.reg.etcd.EtcdRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+
+public class EtcdExample {
+
+ public static void main(String[] args) {
+ // etcd
+ EtcdConfiguration etcdConfig = new
EtcdConfiguration("http://host1:2379,http://host2:2379";, "elasticjob");
+ CoordinatorRegistryCenter registryCenter = new
EtcdRegistryCenter(etcdConfig);
+ registryCenter.init();
+ }
+}
+```
+
+### Spring Boot Starter
+
+```yaml
+elasticjob:
+ reg-center:
+ type: zookeeper # 或 etcd
+ server-lists: host1:2181,host2:2181 # 或
http://host1:2379,http://host2:2379
+ namespace: elasticjob
+```
+
+### Spring Namespace
+
+```xml
+<!-- ZooKeeper -->
+<elasticjob:zookeeper id="regCenter" server-lists="host1:2181,host2:2181"
namespace="elasticjob" />
+
+<!-- etcd -->
+<elasticjob:etcd id="regCenter"
server-lists="http://host1:2379,http://host2:2379"; namespace="elasticjob" />
+```
diff --git
a/docs/content/user-manual/configuration/registry-center/_index.en.md
b/docs/content/user-manual/configuration/registry-center/_index.en.md
new file mode 100644
index 000000000..1c0d7c6c2
--- /dev/null
+++ b/docs/content/user-manual/configuration/registry-center/_index.en.md
@@ -0,0 +1,75 @@
++++
+pre = "<b>4.1.2.1 </b>"
+title = "Registry Center Configuration"
+weight = 1
+chapter = true
++++
+
+ElasticJob supports multiple types of registry centers for coordinating the
scheduling and execution of distributed jobs.
+
+This section describes how to configure different types of registry centers.
+
+## Supported Registry Center Types
+
+| Registry Center Type | Description |
+|---------------------|-------------|
+| [ZooKeeper](/en/user-manual/configuration/registry-center/zookeeper) |
Apache ZooKeeper, distributed coordination service |
+| [etcd](/en/user-manual/configuration/registry-center/etcd) | etcd3,
distributed key-value store |
+
+## Configuration Methods
+
+ElasticJob provides the following methods to configure registry centers:
+
+### Java API
+
+```java
+import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperConfiguration;
+import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+
+public class ZookeeperExample {
+
+ public static void main(String[] args) {
+ // ZooKeeper
+ ZookeeperConfiguration zkConfig = new
ZookeeperConfiguration("host1:2181,host2:2181", "elasticjob");
+ CoordinatorRegistryCenter registryCenter = new
ZookeeperRegistryCenter(zkConfig);
+ registryCenter.init();
+ }
+}
+```
+
+```java
+import org.apache.shardingsphere.elasticjob.reg.etcd.EtcdConfiguration;
+import org.apache.shardingsphere.elasticjob.reg.etcd.EtcdRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+
+public class EtcdExample {
+
+ public static void main(String[] args) {
+ // etcd
+ EtcdConfiguration etcdConfig = new
EtcdConfiguration("http://host1:2379,http://host2:2379";, "elasticjob");
+ CoordinatorRegistryCenter registryCenter = new
EtcdRegistryCenter(etcdConfig);
+ registryCenter.init();
+ }
+}
+```
+
+### Spring Boot Starter
+
+```yaml
+elasticjob:
+ reg-center:
+ type: zookeeper # or etcd
+ server-lists: host1:2181,host2:2181 # or
http://host1:2379,http://host2:2379
+ namespace: elasticjob
+```
+
+### Spring Namespace
+
+```xml
+<!-- ZooKeeper -->
+<elasticjob:zookeeper id="regCenter" server-lists="host1:2181,host2:2181"
namespace="elasticjob" />
+
+<!-- etcd -->
+<elasticjob:etcd id="regCenter"
server-lists="http://host1:2379,http://host2:2379"; namespace="elasticjob" />
+```
diff --git a/docs/content/user-manual/configuration/registry-center/etcd.cn.md
b/docs/content/user-manual/configuration/registry-center/etcd.cn.md
new file mode 100644
index 000000000..ffaa799bc
--- /dev/null
+++ b/docs/content/user-manual/configuration/registry-center/etcd.cn.md
@@ -0,0 +1,98 @@
++++
+title = "etcd 配置"
+weight = 2
++++
+
+etcd 是一个分布式键值存储系统，ElasticJob 支持使用 etcd3 作为注册中心。
+
+## Maven 依赖
+
+```xml
+<dependency>
+ <groupId>org.apache.shardingsphere.elasticjob</groupId>
+ <artifactId>elasticjob-registry-center-etcd</artifactId>
+ <version>${elasticjob.version}</version>
+</dependency>
+```
+
+## 配置项
+
+### 可配置属性
+
+| 属性名 | 类型 | 缺省值 | 描述 |
+|-------------------------------|:-------|:------|:--------------------|
+| serverLists | String | | 连接 etcd 服务器的列表 |
+| namespace | String | | etcd 的命名空间 |
+| connectionTimeoutMilliseconds | long | 5000 | 连接超时毫秒数 |
+| username | String | | 认证用户名 |
+| password | String | | 认证密码 |
+| authority | String | | HTTP/2 的 authority 头 |
+
+### 核心配置项说明
+
+**serverLists:**
+
+包括 IP 地址和端口号，多个地址用逗号分隔，需要包含协议前缀，如: `http://host1:2379,http://host2:2379`
+
+**namespace:**
+
+etcd 的命名空间，用于隔离不同的作业集群。建议使用有意义的名称，如: `elasticjob`
+
+**username 和 password:**
+
+etcd 的认证信息。如果 etcd 启用了认证，需要配置用户名和密码。
+
+**authority:**
+
+HTTP/2 的 authority 头，用于某些特殊的网络环境配置。
+
+## 使用示例
+
+### Java API
+
+```java
+import org.apache.shardingsphere.elasticjob.reg.etcd.EtcdConfiguration;
+import org.apache.shardingsphere.elasticjob.reg.etcd.EtcdRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+
+public class EtcdRegistryCenterExample {
+
+ public static void main(String[] args) {
+ EtcdConfiguration etcdConfig = new
EtcdConfiguration("http://host1:2379,http://host2:2379";, "elasticjob");
+ etcdConfig.setConnectionTimeoutMilliseconds(5000);
+ etcdConfig.setUsername("root");
+ etcdConfig.setPassword("password");
+
+ CoordinatorRegistryCenter registryCenter = new
EtcdRegistryCenter(etcdConfig);
+ registryCenter.init();
+
+ // 使用注册中心...
+
+ registryCenter.close();
+ }
+}
+```
+
+### Spring Boot Starter
+
+```yaml
+elasticjob:
+ reg-center:
+ type: etcd
+ server-lists: http://host1:2379,http://host2:2379
+ namespace: elasticjob
+ connection-timeout-milliseconds: 5000
+ username: root
+ password: password
+```
+
+### Spring Namespace
+
+```xml
+<elasticjob:etcd id="regCenter"
+ server-lists="http://host1:2379,http://host2:2379";
+ namespace="elasticjob"
+ connection-timeout-milliseconds="5000"
+ username="root"
+ password="password" />
+```
diff --git a/docs/content/user-manual/configuration/registry-center/etcd.en.md
b/docs/content/user-manual/configuration/registry-center/etcd.en.md
new file mode 100644
index 000000000..af7ff5270
--- /dev/null
+++ b/docs/content/user-manual/configuration/registry-center/etcd.en.md
@@ -0,0 +1,98 @@
++++
+title = "etcd Configuration"
+weight = 2
++++
+
+etcd is a distributed key-value store system. ElasticJob supports using etcd3
as a registry center.
+
+## Maven Dependency
+
+```xml
+<dependency>
+ <groupId>org.apache.shardingsphere.elasticjob</groupId>
+ <artifactId>elasticjob-registry-center-etcd</artifactId>
+ <version>${elasticjob.version}</version>
+</dependency>
+```
+
+## Configuration
+
+### Configuration Properties
+
+| Name | Data Type | Default Value | Description
|
+|-------------------------------|:----------|:--------------|:---------------------------------------------------------|
+| serverLists | String | | etcd server IP
list |
+| namespace | String | | etcd namespace
|
+| connectionTimeoutMilliseconds | long | 5000 | Connection
timeout in milliseconds |
+| username | String | | Authentication
username |
+| password | String | | Authentication
password |
+| authority | String | | Authority header
for HTTP/2 |
+
+### Core Configuration Description
+
+**serverLists:**
+
+Include IP and port, multiple addresses are separated by commas, protocol
prefix is required, such as: `http://host1:2379,http://host2:2379`
+
+**namespace:**
+
+etcd namespace for isolating different job clusters. It is recommended to use
meaningful names, such as: `elasticjob`
+
+**username and password:**
+
+etcd authentication information. If etcd has authentication enabled, you need
to configure both username and password.
+
+**authority:**
+
+HTTP/2 authority header, used for some special network environment
configurations.
+
+## Usage Examples
+
+### Java API
+
+```java
+import org.apache.shardingsphere.elasticjob.reg.etcd.EtcdConfiguration;
+import org.apache.shardingsphere.elasticjob.reg.etcd.EtcdRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+
+public class EtcdRegistryCenterExample {
+
+ public static void main(String[] args) {
+ EtcdConfiguration etcdConfig = new
EtcdConfiguration("http://host1:2379,http://host2:2379";, "elasticjob");
+ etcdConfig.setConnectionTimeoutMilliseconds(5000);
+ etcdConfig.setUsername("root");
+ etcdConfig.setPassword("password");
+
+ CoordinatorRegistryCenter registryCenter = new
EtcdRegistryCenter(etcdConfig);
+ registryCenter.init();
+
+ // Use registry center...
+
+ registryCenter.close();
+ }
+}
+```
+
+### Spring Boot Starter
+
+```yaml
+elasticjob:
+ reg-center:
+ type: etcd
+ server-lists: http://host1:2379,http://host2:2379
+ namespace: elasticjob
+ connection-timeout-milliseconds: 5000
+ username: root
+ password: password
+```
+
+### Spring Namespace
+
+```xml
+<elasticjob:etcd id="regCenter"
+ server-lists="http://host1:2379,http://host2:2379";
+ namespace="elasticjob"
+ connection-timeout-milliseconds="5000"
+ username="root"
+ password="password" />
+```
diff --git
a/docs/content/user-manual/configuration/registry-center/zookeeper.cn.md
b/docs/content/user-manual/configuration/registry-center/zookeeper.cn.md
new file mode 100644
index 000000000..76fba223e
--- /dev/null
+++ b/docs/content/user-manual/configuration/registry-center/zookeeper.cn.md
@@ -0,0 +1,118 @@
++++
+title = "ZooKeeper 配置"
+weight = 1
++++
+
+Apache ZooKeeper 是 ElasticJob 默认且最成熟的注册中心实现。
+
+## Maven 依赖
+
+```xml
+<dependency>
+ <groupId>org.apache.shardingsphere.elasticjob</groupId>
+ <artifactId>elasticjob-registry-center-zookeeper-curator</artifactId>
+ <version>${elasticjob.version}</version>
+</dependency>
+```
+
+## 配置项
+
+### 可配置属性
+
+| 属性名 | 类型 | 缺省值 | 描述 |
+|-------------------------------|:-------|:------|:--------------------|
+| serverLists | String | | 连接 ZooKeeper 服务器的列表 |
+| namespace | String | | ZooKeeper 的命名空间 |
+| baseSleepTimeMilliseconds | int | 1000 | 等待重试的间隔时间的初始毫秒数 |
+| maxSleepTimeMilliseconds | String | 3000 | 等待重试的间隔时间的最大毫秒数 |
+| maxRetries | String | 3 | 最大重试次数 |
+| sessionTimeoutMilliseconds | int | 60000 | 会话超时毫秒数 |
+| connectionTimeoutMilliseconds | int | 15000 | 连接超时毫秒数 |
+| digest | String | 无需验证 | 连接 ZooKeeper 的权限令牌 |
+| ensembleTracker | boolean | true | 是否监听集群配置变化 |
+
+### 核心配置项说明
+
+**serverLists:**
+
+包括 IP 地址和端口号，多个地址用逗号分隔，如: `host1:2181,host2:2181`
+
+**namespace:**
+
+ZooKeeper 的命名空间，用于隔离不同的作业集群。建议使用有意义的名称，如: `elasticjob`
+
+**digest:**
+
+ZooKeeper 的权限认证信息，格式为 `username:password`。如果 ZooKeeper 启用了 SASL 认证，可以使用此配置项。
+
+**ensembleTracker:**
+
+允许配置是否监听集群配置变化。此配置项用于解决通过虚拟 IP 或负载均衡器连接 ZooKeeper 集群时的问题。
+
+在 Kubernetes 等环境中，Pod 可能随时重启，Pod IP 会变化。如果使用 HA 启用的 ZooKeeper 集群，ZooKeeper
客户端可能在 Ensemble Tracking 期间返回无法解析的 URL，导致连接失败。
+将此配置设置为 `false` 可以让 Curator 始终通过服务集群 IP 连接 ZooKeeper。
+
+> 参考: [Issue
#2072](https://github.com/apache/shardingsphere-elasticjob/issues/2072) - Build
CuratorFrameworkFactory supports an option to skip Ensemble tracking
+
+## 使用示例
+
+### Java API
+
+```java
+import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperConfiguration;
+import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+
+public class ZookeeperRegistryCenterExample {
+
+ public static void main(String[] args) {
+ ZookeeperConfiguration zkConfig = new
ZookeeperConfiguration("host1:2181,host2:2181", "elasticjob");
+ zkConfig.setBaseSleepTimeMilliseconds(1000);
+ zkConfig.setMaxSleepTimeMilliseconds(3000);
+ zkConfig.setMaxRetries(3);
+ zkConfig.setSessionTimeoutMilliseconds(60000);
+ zkConfig.setConnectionTimeoutMilliseconds(15000);
+ zkConfig.setDigest("username:password");
+ zkConfig.setEnsembleTracker(true);
+
+ CoordinatorRegistryCenter registryCenter = new
ZookeeperRegistryCenter(zkConfig);
+ registryCenter.init();
+
+ // 使用注册中心...
+
+ registryCenter.close();
+ }
+}
+```
+
+### Spring Boot Starter
+
+```yaml
+elasticjob:
+ reg-center:
+ type: zookeeper
+ server-lists: host1:2181,host2:2181
+ namespace: elasticjob
+ base-sleep-time-milliseconds: 1000
+ max-sleep-time-milliseconds: 3000
+ max-retries: 3
+ session-timeout-milliseconds: 60000
+ connection-timeout-milliseconds: 15000
+ digest: username:password
+ ensemble-tracker: true
+```
+
+### Spring Namespace
+
+```xml
+<elasticjob:zookeeper id="regCenter"
+ server-lists="host1:2181,host2:2181"
+ namespace="elasticjob"
+ base-sleep-time-milliseconds="1000"
+ max-sleep-time-milliseconds="3000"
+ max-retries="3"
+ session-timeout-milliseconds="60000"
+ connection-timeout-milliseconds="15000"
+ digest="username:password"
+ ensemble-tracker="true" />
+```
diff --git
a/docs/content/user-manual/configuration/registry-center/zookeeper.en.md
b/docs/content/user-manual/configuration/registry-center/zookeeper.en.md
new file mode 100644
index 000000000..b400f4445
--- /dev/null
+++ b/docs/content/user-manual/configuration/registry-center/zookeeper.en.md
@@ -0,0 +1,118 @@
++++
+title = "ZooKeeper Configuration"
+weight = 1
++++
+
+Apache ZooKeeper is the default and most mature registry center implementation
for ElasticJob.
+
+## Maven Dependency
+
+```xml
+<dependency>
+ <groupId>org.apache.shardingsphere.elasticjob</groupId>
+ <artifactId>elasticjob-registry-center-zookeeper-curator</artifactId>
+ <version>${elasticjob.version}</version>
+</dependency>
+```
+
+## Configuration
+
+### Configuration Properties
+
+| Name | Data Type | Default Value | Description
|
+|-------------------------------|:----------|:--------------|:---------------------------------------------------------|
+| serverLists | String | | ZooKeeper server
IP list |
+| namespace | String | | ZooKeeper
namespace |
+| baseSleepTimeMilliseconds | int | 1000 | The initial
value of milliseconds for the retry interval |
+| maxSleepTimeMilliseconds | String | 3000 | The maximum
value of milliseconds for the retry interval |
+| maxRetries | String | 3 | Maximum number
of retries |
+| sessionTimeoutMilliseconds | int | 60000 | Session timeout
in milliseconds |
+| connectionTimeoutMilliseconds | int | 15000 | Connection
timeout in milliseconds |
+| digest | String | no need | Permission token
to connect to ZooKeeper |
+| ensembleTracker | boolean | true | Whether to watch
ensemble configuration changes |
+
+### Core Configuration Description
+
+**serverLists:**
+
+Include IP and port, multiple addresses are separated by commas, such as:
`host1:2181,host2:2181`
+
+**namespace:**
+
+ZooKeeper namespace for isolating different job clusters. It is recommended to
use meaningful names, such as: `elasticjob`
+
+**digest:**
+
+ZooKeeper authentication information, format is `username:password`. If
ZooKeeper has SASL authentication enabled, you can use this configuration.
+
+**ensembleTracker:**
+
+Allows configuring whether to watch ensemble configuration changes. This
configuration is used to solve the problem when connecting to ZooKeeper
clusters via Virtual IPs or load balancers.
+
+In environments like Kubernetes, Pods may restart at any time and Pod IPs will
change. If using HA enabled Zookeeper clusters, the Zookeeper Client may return
unresolvable URLs during Ensemble Tracking, causing connection failures.
+Setting this configuration to `false` allows Curator to always connect to
ZooKeeper via the service cluster IP.
+
+> Reference: [Issue
#2072](https://github.com/apache/shardingsphere-elasticjob/issues/2072) - Build
CuratorFrameworkFactory supports an option to skip Ensemble tracking
+
+## Usage Examples
+
+### Java API
+
+```java
+import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperConfiguration;
+import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+
+public class ZookeeperRegistryCenterExample {
+
+ public static void main(String[] args) {
+ ZookeeperConfiguration zkConfig = new
ZookeeperConfiguration("host1:2181,host2:2181", "elasticjob");
+ zkConfig.setBaseSleepTimeMilliseconds(1000);
+ zkConfig.setMaxSleepTimeMilliseconds(3000);
+ zkConfig.setMaxRetries(3);
+ zkConfig.setSessionTimeoutMilliseconds(60000);
+ zkConfig.setConnectionTimeoutMilliseconds(15000);
+ zkConfig.setDigest("username:password");
+ zkConfig.setEnsembleTracker(true);
+
+ CoordinatorRegistryCenter registryCenter = new
ZookeeperRegistryCenter(zkConfig);
+ registryCenter.init();
+
+ // Use registry center...
+
+ registryCenter.close();
+ }
+}
+```
+
+### Spring Boot Starter
+
+```yaml
+elasticjob:
+ reg-center:
+ type: zookeeper
+ server-lists: host1:2181,host2:2181
+ namespace: elasticjob
+ base-sleep-time-milliseconds: 1000
+ max-sleep-time-milliseconds: 3000
+ max-retries: 3
+ session-timeout-milliseconds: 60000
+ connection-timeout-milliseconds: 15000
+ digest: username:password
+ ensemble-tracker: true
+```
+
+### Spring Namespace
+
+```xml
+<elasticjob:zookeeper id="regCenter"
+ server-lists="host1:2181,host2:2181"
+ namespace="elasticjob"
+ base-sleep-time-milliseconds="1000"
+ max-sleep-time-milliseconds="3000"
+ max-retries="3"
+ session-timeout-milliseconds="60000"
+ connection-timeout-milliseconds="15000"
+ digest="username:password"
+ ensemble-tracker="true" />
+```
diff --git a/lifecycle/pom.xml b/lifecycle/pom.xml
index 8fb33eff1..5bfcb8532 100644
--- a/lifecycle/pom.xml
+++ b/lifecycle/pom.xml
@@ -38,6 +38,12 @@
<version>${project.parent.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.shardingsphere.elasticjob</groupId>
+ <artifactId>elasticjob-registry-center-etcd</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.shardingsphere.elasticjob</groupId>
diff --git
a/lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lifecycle/internal/reg/RegistryCenterFactory.java
b/lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lifecycle/internal/reg/RegistryCenterFactory.java
index dc63c17fb..327d0b59f 100644
---
a/lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lifecycle/internal/reg/RegistryCenterFactory.java
+++
b/lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lifecycle/internal/reg/RegistryCenterFactory.java
@@ -24,11 +24,11 @@ import com.google.common.hash.Hashing;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
-import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperConfiguration;
-import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.spi.RegistryCenterCreator;
import java.nio.charset.StandardCharsets;
import java.util.Map;
+import java.util.ServiceLoader;
import java.util.concurrent.ConcurrentHashMap;
/**
@@ -40,6 +40,8 @@ public final class RegistryCenterFactory {
private static final Map<HashCode, CoordinatorRegistryCenter>
REG_CENTER_REGISTRY = new ConcurrentHashMap<>();
+ private static final ServiceLoader<RegistryCenterCreator> CREATOR_LOADER =
ServiceLoader.load(RegistryCenterCreator.class);
+
/**
* Create a {@link CoordinatorRegistryCenter} or return the existing one
if there is one set up with the same {@code connectionString}, {@code
namespace} and {@code digest} already.
*
@@ -64,10 +66,11 @@ public final class RegistryCenterFactory {
private static CoordinatorRegistryCenter
newCoordinatorRegistryCenter(final String connectString,
final String namespace,
final String digest) {
- final ZookeeperConfiguration zkConfig = new
ZookeeperConfiguration(connectString, namespace);
- if (!Strings.isNullOrEmpty(digest)) {
- zkConfig.setDigest(digest);
+ for (RegistryCenterCreator creator : CREATOR_LOADER) {
+ if (creator.supports(connectString)) {
+ return creator.create(connectString, namespace, digest);
+ }
}
- return new ZookeeperRegistryCenter(zkConfig);
+ throw new IllegalArgumentException("No registry center creator found
for connect string: " + connectString);
}
}
diff --git a/pom.xml b/pom.xml
index 8e9321ae9..8c95ea1e6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -73,6 +73,7 @@
<zookeeper.version>3.9.5</zookeeper.version>
<curator.version>5.9.0</curator.version>
+ <jetcd.version>0.7.7</jetcd.version>
<mail.version>1.6.0</mail.version>
<commons-codec.version>1.16.0</commons-codec.version>
@@ -221,6 +222,12 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>io.etcd</groupId>
+ <artifactId>jetcd-core</artifactId>
+ <version>${jetcd.version}</version>
+ </dependency>
+
<dependency>
<groupId>com.sun.mail</groupId>
<artifactId>javax.mail</artifactId>
diff --git
a/registry-center/api/src/main/java/org/apache/shardingsphere/elasticjob/reg/spi/RegistryCenterCreator.java
b/registry-center/api/src/main/java/org/apache/shardingsphere/elasticjob/reg/spi/RegistryCenterCreator.java
new file mode 100644
index 000000000..6f0aec7ff
--- /dev/null
+++
b/registry-center/api/src/main/java/org/apache/shardingsphere/elasticjob/reg/spi/RegistryCenterCreator.java
@@ -0,0 +1,46 @@
+/*
+ * 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.elasticjob.reg.spi;
+
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+
+/**
+ * Registry center creator.
+ *
+ * <p>Implementations of this interface are used to create different types of
registry centers.</p>
+ */
+public interface RegistryCenterCreator {
+
+ /**
+ * Check if this creator supports the given connection string.
+ *
+ * @param connectString connection string
+ * @return true if supported, false otherwise
+ */
+ boolean supports(String connectString);
+
+ /**
+ * Create a coordinator registry center.
+ *
+ * @param connectString connection string
+ * @param namespace namespace
+ * @param digest authentication information
+ * @return coordinator registry center
+ */
+ CoordinatorRegistryCenter create(String connectString, String namespace,
String digest);
+}
diff --git a/lifecycle/pom.xml b/registry-center/provider/etcd/pom.xml
similarity index 76%
copy from lifecycle/pom.xml
copy to registry-center/provider/etcd/pom.xml
index 8fb33eff1..2bd8d78df 100644
--- a/lifecycle/pom.xml
+++ b/registry-center/provider/etcd/pom.xml
@@ -6,7 +6,7 @@
~ 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
@@ -20,35 +20,40 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.shardingsphere.elasticjob</groupId>
- <artifactId>elasticjob</artifactId>
+ <artifactId>elasticjob-regitry-center-provider</artifactId>
<version>3.0.6-SNAPSHOT</version>
</parent>
- <artifactId>elasticjob-lifecycle</artifactId>
+ <artifactId>elasticjob-registry-center-etcd</artifactId>
<name>${project.artifactId}</name>
<dependencies>
<dependency>
<groupId>org.apache.shardingsphere.elasticjob</groupId>
- <artifactId>elasticjob-bootstrap</artifactId>
+ <artifactId>elasticjob-registry-center-api</artifactId>
<version>${project.parent.version}</version>
</dependency>
+
<dependency>
- <groupId>org.apache.shardingsphere.elasticjob</groupId>
-
<artifactId>elasticjob-registry-center-zookeeper-curator</artifactId>
- <version>${project.parent.version}</version>
- <scope>provided</scope>
+ <groupId>io.etcd</groupId>
+ <artifactId>jetcd-core</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.shardingsphere.elasticjob</groupId>
- <artifactId>elasticjob-test-util</artifactId>
- <version>${project.parent.version}</version>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
- <groupId>commons-codec</groupId>
- <artifactId>commons-codec</artifactId>
+ <groupId>org.apache.shardingsphere.elasticjob</groupId>
+ <artifactId>elasticjob-test-util</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
</dependency>
</dependencies>
</project>
diff --git
a/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdConfiguration.java
b/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdConfiguration.java
new file mode 100644
index 000000000..6c28d540f
--- /dev/null
+++
b/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdConfiguration.java
@@ -0,0 +1,73 @@
+/*
+ * 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.elasticjob.reg.etcd;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+
+/**
+ * Etcd configuration.
+ */
+@Getter
+@Setter
+@RequiredArgsConstructor
+public final class EtcdConfiguration {
+
+ /**
+ * Server list of etcd.
+ *
+ * <p>
+ * Include IP addresses and ports,
+ * Multiple IP address split by comma.
+ * For example: http://host1:2379,http://host2:2379
+ * </p>
+ */
+ @SuppressWarnings("JavadocLinkAsPlainText")
+ private final String serverLists;
+
+ /**
+ * Namespace.
+ */
+ private final String namespace;
+
+ /**
+ * Connection timeout milliseconds.
+ */
+ private long connectionTimeoutMilliseconds = 5000;
+
+ /**
+ * Username for authentication.
+ */
+ private String username;
+
+ /**
+ * Password for authentication.
+ */
+ private String password;
+
+ /**
+ * Whether to use HTTPs.
+ */
+ private boolean ssl;
+
+ /**
+ * Authority header for HTTP/2.
+ */
+ private String authority;
+}
diff --git
a/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdRegistryCenter.java
b/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdRegistryCenter.java
new file mode 100644
index 000000000..2c676275f
--- /dev/null
+++
b/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdRegistryCenter.java
@@ -0,0 +1,481 @@
+/*
+ * 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.elasticjob.reg.etcd;
+
+import com.google.common.base.Strings;
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.ClientBuilder;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Lock;
+import io.etcd.jetcd.Watch;
+import io.etcd.jetcd.Watch.Watcher;
+import io.etcd.jetcd.kv.GetResponse;
+import io.etcd.jetcd.lease.LeaseGrantResponse;
+import io.etcd.jetcd.lock.LockResponse;
+import io.etcd.jetcd.options.DeleteOption;
+import io.etcd.jetcd.options.GetOption;
+import io.etcd.jetcd.options.PutOption;
+import io.etcd.jetcd.options.WatchOption;
+import io.etcd.jetcd.watch.WatchEvent;
+import io.etcd.jetcd.watch.WatchResponse;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.base.LeaderExecutionCallback;
+import
org.apache.shardingsphere.elasticjob.reg.base.transaction.TransactionOperation;
+import org.apache.shardingsphere.elasticjob.reg.exception.RegException;
+import org.apache.shardingsphere.elasticjob.reg.exception.RegExceptionHandler;
+import
org.apache.shardingsphere.elasticjob.reg.listener.ConnectionStateChangedEventListener;
+import org.apache.shardingsphere.elasticjob.reg.listener.DataChangedEvent;
+import org.apache.shardingsphere.elasticjob.reg.listener.DataChangedEvent.Type;
+import
org.apache.shardingsphere.elasticjob.reg.listener.DataChangedEventListener;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Registry center of etcd.
+ */
+@Slf4j
+public final class EtcdRegistryCenter implements CoordinatorRegistryCenter {
+
+ @Getter(AccessLevel.PROTECTED)
+ private final EtcdConfiguration etcdConfig;
+
+ private final Map<String, ByteSequence> cache = new ConcurrentHashMap<>();
+
+ private final Map<String, List<Watcher>> watches = new
ConcurrentHashMap<>();
+
+ private final Map<String, List<ConnectionStateChangedEventListener>>
connStateListeners = new ConcurrentHashMap<>();
+
+ private final Map<Long, Long> leaseIdMap = new ConcurrentHashMap<>();
+
+ private final AtomicLong sequenceCounter = new AtomicLong(0);
+
+ @Getter
+ private Client client;
+
+ private KV kvClient;
+
+ private Lease leaseClient;
+
+ private Lock lockClient;
+
+ public EtcdRegistryCenter(final EtcdConfiguration etcdConfig) {
+ this.etcdConfig = etcdConfig;
+ }
+
+ @Override
+ public void init() {
+ log.debug("Elastic job: etcd registry center init, server lists is:
{}.", etcdConfig.getServerLists());
+ try {
+ ClientBuilder builder = Client.builder()
+ .endpoints(etcdConfig.getServerLists().split(","))
+
.connectTimeout(Duration.ofMillis(etcdConfig.getConnectionTimeoutMilliseconds()));
+ if (!Strings.isNullOrEmpty(etcdConfig.getUsername()) &&
!Strings.isNullOrEmpty(etcdConfig.getPassword())) {
+ builder.user(ByteSequence.from(etcdConfig.getUsername(),
StandardCharsets.UTF_8));
+ builder.password(ByteSequence.from(etcdConfig.getPassword(),
StandardCharsets.UTF_8));
+ }
+ if (!Strings.isNullOrEmpty(etcdConfig.getAuthority())) {
+ builder.authority(etcdConfig.getAuthority());
+ }
+ client = builder.build();
+ kvClient = client.getKVClient();
+ leaseClient = client.getLeaseClient();
+ lockClient = client.getLockClient();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ }
+ }
+
+ @Override
+ public void close() {
+ for (Long leaseId : leaseIdMap.values()) {
+ try {
+ leaseClient.revoke(leaseId).get();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ log.warn("Failed to revoke lease: {}", leaseId, ex);
+ }
+ }
+ leaseIdMap.clear();
+ cache.clear();
+ for (List<Watcher> watchList : watches.values()) {
+ watchList.forEach(Watcher::close);
+ }
+ watches.clear();
+ if (client != null) {
+ client.close();
+ }
+ }
+
+ @Override
+ public String get(final String key) {
+ ByteSequence cachedValue = cache.get(key);
+ if (cachedValue != null) {
+ return cachedValue.toString(StandardCharsets.UTF_8);
+ }
+ return getDirectly(key);
+ }
+
+ @Override
+ public String getDirectly(final String key) {
+ try {
+ GetResponse response = kvClient.get(toByteSequence(key)).get();
+ if (response.getKvs().isEmpty()) {
+ return null;
+ }
+ return
response.getKvs().get(0).getValue().toString(StandardCharsets.UTF_8);
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ return null;
+ }
+ }
+
+ @Override
+ public List<String> getChildrenKeys(final String key) {
+ try {
+ String prefix = key.endsWith("/") ? key : key + "/";
+ GetOption option = GetOption.newBuilder()
+ .withPrefix(toByteSequence(prefix))
+ .withSortField(GetOption.SortTarget.KEY)
+ .withSortOrder(GetOption.SortOrder.DESCEND)
+ .build();
+ GetResponse response = kvClient.get(toByteSequence(prefix),
option).get();
+ List<String> result = new ArrayList<>();
+ for (KeyValue kv : response.getKvs()) {
+ String childKey = kv.getKey().toString(StandardCharsets.UTF_8);
+ String relativeKey = childKey.substring(prefix.length());
+ if (!relativeKey.isEmpty() && !relativeKey.contains("/")) {
+ result.add(relativeKey);
+ }
+ }
+ result.sort(Comparator.reverseOrder());
+ return result;
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ return Collections.emptyList();
+ }
+ }
+
+ @Override
+ public int getNumChildren(final String key) {
+ return getChildrenKeys(key).size();
+ }
+
+ @Override
+ public boolean isExisted(final String key) {
+ try {
+ GetResponse response = kvClient.get(toByteSequence(key)).get();
+ return !response.getKvs().isEmpty();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ return false;
+ }
+ }
+
+ @Override
+ public void persist(final String key, final String value) {
+ try {
+ kvClient.put(toByteSequence(key), toByteSequence(value)).get();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ }
+ }
+
+ @Override
+ public void update(final String key, final String value) {
+ try {
+ kvClient.put(toByteSequence(key), toByteSequence(value)).get();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ }
+ }
+
+ @Override
+ public void persistEphemeral(final String key, final String value) {
+ try {
+ long leaseId = createLease(30);
+ kvClient.put(toByteSequence(key), toByteSequence(value),
PutOption.newBuilder().withLeaseId(leaseId).build()).get();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ }
+ }
+
+ @Override
+ public String persistSequential(final String key, final String value) {
+ try {
+ long seq = sequenceCounter.incrementAndGet();
+ String sequentialKey = key + String.format("%010d", seq);
+ kvClient.put(toByteSequence(sequentialKey),
toByteSequence(value)).get();
+ return sequentialKey;
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ return null;
+ }
+ }
+
+ @Override
+ public void persistEphemeralSequential(final String key) {
+ try {
+ long leaseId = createLease(30);
+ long seq = sequenceCounter.incrementAndGet();
+ String sequentialKey = key + String.format("%010d", seq);
+ kvClient.put(toByteSequence(sequentialKey), toByteSequence(""),
PutOption.newBuilder().withLeaseId(leaseId).build()).get();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ }
+ }
+
+ @Override
+ public void remove(final String key) {
+ try {
+ kvClient.delete(toByteSequence(key),
DeleteOption.newBuilder().withPrefix(toByteSequence(key)).build()).get();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ }
+ }
+
+ @Override
+ public long getRegistryCenterTime(final String key) {
+ try {
+ persist(key, "");
+ GetResponse response = kvClient.get(toByteSequence(key)).get();
+ if (!response.getKvs().isEmpty()) {
+ return response.getKvs().get(0).getModRevision();
+ }
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ }
+ throw new RegException(new IllegalStateException("Cannot get registry
center time."));
+ }
+
+ @Override
+ public Object getRawClient() {
+ return client;
+ }
+
+ @Override
+ public void executeInLeader(final String key, final
LeaderExecutionCallback callback) {
+ try {
+ ByteSequence lockKey = toByteSequence(key);
+ LockResponse lockResponse = lockClient.lock(lockKey, 30).get();
+ try {
+ callback.execute();
+ } finally {
+ lockClient.unlock(lockResponse.getKey()).get();
+ }
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ handleException(ex);
+ }
+ }
+
+ @Override
+ public void executeInTransaction(final List<TransactionOperation>
transactionOperations) throws Exception {
+ for (TransactionOperation operation : transactionOperations) {
+ switch (operation.getType()) {
+ case CHECK_EXISTS:
+ if (!isExisted(operation.getKey())) {
+ throw new RegException(new IllegalStateException("Key
does not exist: " + operation.getKey()));
+ }
+ break;
+ case ADD:
+ persist(operation.getKey(), operation.getValue());
+ break;
+ case UPDATE:
+ update(operation.getKey(), operation.getValue());
+ break;
+ case DELETE:
+ remove(operation.getKey());
+ break;
+ default:
+ throw new
UnsupportedOperationException(operation.toString());
+ }
+ }
+ }
+
+ @Override
+ public void addCacheData(final String cachePath) {
+ try {
+ String prefix = cachePath.endsWith("/") ? cachePath : cachePath +
"/";
+ GetOption option = GetOption.newBuilder()
+ .withPrefix(toByteSequence(prefix))
+ .build();
+ GetResponse response = kvClient.get(toByteSequence(prefix),
option).get();
+ for (KeyValue kv : response.getKvs()) {
+ cache.put(kv.getKey().toString(StandardCharsets.UTF_8),
kv.getValue());
+ }
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ RegExceptionHandler.handleException(ex);
+ }
+ }
+
+ @Override
+ public void evictCacheData(final String cachePath) {
+ String prefix = cachePath.endsWith("/") ? cachePath : cachePath + "/";
+ cache.entrySet().removeIf(entry -> entry.getKey().startsWith(prefix));
+ }
+
+ @Override
+ public Object getRawCache(final String cachePath) {
+ return cache;
+ }
+
+ @Override
+ public void watch(final String key, final DataChangedEventListener
listener, final Executor executor) {
+ String prefix = key.endsWith("/") ? key : key + "/";
+ WatchOption option = WatchOption.newBuilder()
+ .withPrefix(toByteSequence(prefix))
+ .build();
+ Watcher watcher =
client.getWatchClient().watch(toByteSequence(prefix), option, new
EtcdWatchListener(key, listener, executor));
+ watches.computeIfAbsent(key, k -> new ArrayList<>()).add(watcher);
+ }
+
+ @Override
+ public void removeDataListeners(final String key) {
+ List<Watcher> watchList = watches.remove(key);
+ if (watchList != null) {
+ watchList.forEach(Watcher::close);
+ }
+ }
+
+ @Override
+ public void addConnectionStateChangedEventListener(final String key,
+ final
ConnectionStateChangedEventListener listener) {
+ connStateListeners.computeIfAbsent(key, k -> new
ArrayList<>()).add(listener);
+ }
+
+ @Override
+ public void removeConnStateListener(final String key) {
+ connStateListeners.remove(key);
+ }
+
+ private long createLease(final long ttlSeconds) {
+ try {
+ LeaseGrantResponse response = leaseClient.grant(ttlSeconds).get();
+ long leaseId = response.getID();
+ leaseIdMap.put(leaseId, leaseId);
+ return leaseId;
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ throw new RegException(ex);
+ }
+ }
+
+ private ByteSequence toByteSequence(final String value) {
+ return ByteSequence.from(value, StandardCharsets.UTF_8);
+ }
+
+ private Type getTypeFromWatchEvent(final WatchEvent event) {
+ switch (event.getEventType()) {
+ case PUT:
+ return event.getPrevKV().getValue().isEmpty() ? Type.ADDED :
Type.UPDATED;
+ case DELETE:
+ return Type.DELETED;
+ default:
+ return Type.IGNORED;
+ }
+ }
+
+ private void handleException(final Exception ex) {
+ if (ex instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ } else {
+ throw new RegException(ex);
+ }
+ }
+
+ private class EtcdWatchListener implements Watch.Listener {
+
+ private final String key;
+
+ private final DataChangedEventListener listener;
+
+ private final Executor executor;
+
+ EtcdWatchListener(final String key, final DataChangedEventListener
listener, final Executor executor) {
+ this.key = key;
+ this.listener = listener;
+ this.executor = executor;
+ }
+
+ @Override
+ public void onNext(final WatchResponse response) {
+ for (WatchEvent event : response.getEvents()) {
+ Type type = getTypeFromWatchEvent(event);
+ String eventKey =
event.getKeyValue().getKey().toString(StandardCharsets.UTF_8);
+ String eventValue =
event.getKeyValue().getValue().toString(StandardCharsets.UTF_8);
+ if (executor != null) {
+ executor.execute(() -> listener.onChange(new
DataChangedEvent(type, eventKey, eventValue)));
+ } else {
+ listener.onChange(new DataChangedEvent(type, eventKey,
eventValue));
+ }
+ }
+ }
+
+ @Override
+ public void onError(final Throwable throwable) {
+ log.error("Watch error for key: {}", key, throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ log.debug("Watch completed for key: {}", key);
+ }
+ }
+}
diff --git
a/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdRegistryCenterCreator.java
b/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdRegistryCenterCreator.java
new file mode 100644
index 000000000..9485e0244
--- /dev/null
+++
b/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdRegistryCenterCreator.java
@@ -0,0 +1,46 @@
+/*
+ * 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.elasticjob.reg.etcd;
+
+import com.google.common.base.Strings;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.spi.RegistryCenterCreator;
+
+/**
+ * etcd registry center creator.
+ */
+public final class EtcdRegistryCenterCreator implements RegistryCenterCreator {
+
+ @Override
+ public boolean supports(final String connectString) {
+ return connectString.startsWith("http://";) ||
connectString.startsWith("https://";);
+ }
+
+ @Override
+ public CoordinatorRegistryCenter create(final String connectString, final
String namespace, final String digest) {
+ EtcdConfiguration etcdConfig = new EtcdConfiguration(connectString,
namespace);
+ if (!Strings.isNullOrEmpty(digest)) {
+ String[] parts = digest.split(":");
+ if (parts.length == 2) {
+ etcdConfig.setUsername(parts[0]);
+ etcdConfig.setPassword(parts[1]);
+ }
+ }
+ return new EtcdRegistryCenter(etcdConfig);
+ }
+}
diff --git
a/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/exception/EtcdIgnoredExceptionProvider.java
b/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/exception/EtcdIgnoredExceptionProvider.java
new file mode 100644
index 000000000..19ef0d61b
--- /dev/null
+++
b/registry-center/provider/etcd/src/main/java/org/apache/shardingsphere/elasticjob/reg/etcd/exception/EtcdIgnoredExceptionProvider.java
@@ -0,0 +1,36 @@
+/*
+ * 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.elasticjob.reg.etcd.exception;
+
+import io.etcd.jetcd.common.exception.EtcdException;
+import
org.apache.shardingsphere.elasticjob.reg.exception.IgnoredExceptionProvider;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * Etcd ignored exception provider.
+ */
+public final class EtcdIgnoredExceptionProvider implements
IgnoredExceptionProvider {
+
+ @Override
+ public Collection<Class<? extends Throwable>> getIgnoredExceptions() {
+ return Arrays.asList(EtcdException.class, ExecutionException.class);
+ }
+}
diff --git
a/registry-center/provider/etcd/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.reg.exception.IgnoredExceptionProvider
b/registry-center/provider/etcd/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.reg.exception.IgnoredExceptionProvider
new file mode 100644
index 000000000..0ad985920
--- /dev/null
+++
b/registry-center/provider/etcd/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.reg.exception.IgnoredExceptionProvider
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+org.apache.shardingsphere.elasticjob.reg.etcd.exception.EtcdIgnoredExceptionProvider
diff --git
a/registry-center/provider/etcd/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.reg.spi.RegistryCenterCreator
b/registry-center/provider/etcd/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.reg.spi.RegistryCenterCreator
new file mode 100644
index 000000000..5fb354dab
--- /dev/null
+++
b/registry-center/provider/etcd/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.reg.spi.RegistryCenterCreator
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+org.apache.shardingsphere.elasticjob.reg.etcd.EtcdRegistryCenterCreator
diff --git
a/registry-center/provider/etcd/src/test/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdConfigurationTest.java
b/registry-center/provider/etcd/src/test/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdConfigurationTest.java
new file mode 100644
index 000000000..ab113e57d
--- /dev/null
+++
b/registry-center/provider/etcd/src/test/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdConfigurationTest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.elasticjob.reg.etcd;
+
+import org.apache.curator.test.InstanceSpec;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+class EtcdConfigurationTest {
+
+ @Test
+ void assertNewEtcdConfigurationForServerListsAndNamespace() {
+ int randomPort = InstanceSpec.getRandomPort();
+ EtcdConfiguration etcdConfig = new
EtcdConfiguration("http://localhost:"; + randomPort, "myNamespace");
+ assertThat(etcdConfig.getServerLists(), is("http://localhost:"; +
randomPort));
+ assertThat(etcdConfig.getNamespace(), is("myNamespace"));
+ assertThat(etcdConfig.getConnectionTimeoutMilliseconds(), is(5000L));
+ assertThat(etcdConfig.getUsername(), is(nullValue()));
+ assertThat(etcdConfig.getPassword(), is(nullValue()));
+ assertThat(etcdConfig.getAuthority(), is(nullValue()));
+ }
+
+ @Test
+ void assertSetConnectionTimeoutMilliseconds() {
+ EtcdConfiguration etcdConfig = new
EtcdConfiguration("http://localhost:"; + InstanceSpec.getRandomPort(),
"myNamespace");
+ etcdConfig.setConnectionTimeoutMilliseconds(10000L);
+ assertThat(etcdConfig.getConnectionTimeoutMilliseconds(), is(10000L));
+ }
+
+ @Test
+ void assertSetUsernameAndPassword() {
+ EtcdConfiguration etcdConfig = new
EtcdConfiguration("http://localhost:"; + InstanceSpec.getRandomPort(),
"myNamespace");
+ etcdConfig.setUsername("root");
+ etcdConfig.setPassword("password");
+ assertThat(etcdConfig.getUsername(), is("root"));
+ assertThat(etcdConfig.getPassword(), is("password"));
+ }
+
+ @Test
+ void assertSetAuthority() {
+ EtcdConfiguration etcdConfig = new
EtcdConfiguration("http://localhost:"; + InstanceSpec.getRandomPort(),
"myNamespace");
+ etcdConfig.setAuthority("localhost:2379");
+ assertThat(etcdConfig.getAuthority(), is("localhost:2379"));
+ }
+}
diff --git
a/registry-center/provider/etcd/src/test/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdRegistryCenterCreatorTest.java
b/registry-center/provider/etcd/src/test/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdRegistryCenterCreatorTest.java
new file mode 100644
index 000000000..648149a38
--- /dev/null
+++
b/registry-center/provider/etcd/src/test/java/org/apache/shardingsphere/elasticjob/reg/etcd/EtcdRegistryCenterCreatorTest.java
@@ -0,0 +1,49 @@
+/*
+ * 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.elasticjob.reg.etcd;
+
+import org.apache.curator.test.InstanceSpec;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+class EtcdRegistryCenterCreatorTest {
+
+ private final EtcdRegistryCenterCreator creator = new
EtcdRegistryCenterCreator();
+
+ @Test
+ void assertSupportsHttpUrl() {
+ assertThat(creator.supports("http://localhost:"; +
InstanceSpec.getRandomPort()), is(true));
+ }
+
+ @Test
+ void assertSupportsHttpsUrl() {
+ assertThat(creator.supports("https://localhost:"; +
InstanceSpec.getRandomPort()), is(true));
+ }
+
+ @Test
+ void assertNotSupportsZookeeperConnectString() {
+ assertThat(creator.supports("localhost:" +
InstanceSpec.getRandomPort()), is(false));
+ }
+
+ @Test
+ void assertNotSupportsMultipleZookeeperConnectString() {
+ assertThat(creator.supports("host1:2181,host2:2181"), is(false));
+ }
+}
diff --git
a/registry-center/provider/etcd/src/test/java/org/apache/shardingsphere/elasticjob/reg/etcd/exception/EtcdIgnoredExceptionProviderTest.java
b/registry-center/provider/etcd/src/test/java/org/apache/shardingsphere/elasticjob/reg/etcd/exception/EtcdIgnoredExceptionProviderTest.java
new file mode 100644
index 000000000..5f4660009
--- /dev/null
+++
b/registry-center/provider/etcd/src/test/java/org/apache/shardingsphere/elasticjob/reg/etcd/exception/EtcdIgnoredExceptionProviderTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.elasticjob.reg.etcd.exception;
+
+import io.etcd.jetcd.common.exception.EtcdException;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collection;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class EtcdIgnoredExceptionProviderTest {
+
+ private final EtcdIgnoredExceptionProvider provider = new
EtcdIgnoredExceptionProvider();
+
+ @Test
+ void assertGetIgnoredExceptions() {
+ Collection<Class<? extends Throwable>> ignoredExceptions =
provider.getIgnoredExceptions();
+ assertThat(ignoredExceptions.size(), is(2));
+ assertTrue(ignoredExceptions.contains(EtcdException.class));
+
assertTrue(ignoredExceptions.contains(java.util.concurrent.ExecutionException.class));
+ }
+}
diff --git a/registry-center/provider/pom.xml b/registry-center/provider/pom.xml
index 860f15798..70e4af0af 100644
--- a/registry-center/provider/pom.xml
+++ b/registry-center/provider/pom.xml
@@ -29,5 +29,6 @@
<modules>
<module>zookeeper-curator</module>
+ <module>etcd</module>
</modules>
</project>
diff --git
a/registry-center/provider/zookeeper-curator/src/main/java/org/apache/shardingsphere/elasticjob/reg/zookeeper/ZookeeperRegistryCenterCreator.java
b/registry-center/provider/zookeeper-curator/src/main/java/org/apache/shardingsphere/elasticjob/reg/zookeeper/ZookeeperRegistryCenterCreator.java
new file mode 100644
index 000000000..5cb901a47
--- /dev/null
+++
b/registry-center/provider/zookeeper-curator/src/main/java/org/apache/shardingsphere/elasticjob/reg/zookeeper/ZookeeperRegistryCenterCreator.java
@@ -0,0 +1,42 @@
+/*
+ * 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.elasticjob.reg.zookeeper;
+
+import com.google.common.base.Strings;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+import org.apache.shardingsphere.elasticjob.reg.spi.RegistryCenterCreator;
+
+/**
+ * ZooKeeper registry center creator.
+ */
+public final class ZookeeperRegistryCenterCreator implements
RegistryCenterCreator {
+
+ @Override
+ public boolean supports(final String connectString) {
+ return !connectString.startsWith("http://";) &&
!connectString.startsWith("https://";);
+ }
+
+ @Override
+ public CoordinatorRegistryCenter create(final String connectString, final
String namespace, final String digest) {
+ ZookeeperConfiguration zkConfig = new
ZookeeperConfiguration(connectString, namespace);
+ if (!Strings.isNullOrEmpty(digest)) {
+ zkConfig.setDigest(digest);
+ }
+ return new ZookeeperRegistryCenter(zkConfig);
+ }
+}
diff --git
a/registry-center/provider/zookeeper-curator/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.reg.spi.RegistryCenterCreator
b/registry-center/provider/zookeeper-curator/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.reg.spi.RegistryCenterCreator
new file mode 100644
index 000000000..12d26699c
--- /dev/null
+++
b/registry-center/provider/zookeeper-curator/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.reg.spi.RegistryCenterCreator
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenterCreator
