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.git
The following commit(s) were added to refs/heads/master by this push:
new 8b8b88d6f93 Refactor coordinator registry center creation with SPI
(#39771)
8b8b88d6f93 is described below
commit 8b8b88d6f932f7d311a3f477c4d08fdda61a7789
Author: Liang Zhang <[email protected]>
AuthorDate: Wed Sep 9 00:55:49 2026 +0800
Refactor coordinator registry center creation with SPI (#39771)
* Simplify altered encrypt rule cleanup
* Refactor coordinator registry center creation with SPI
---
kernel/data-pipeline/core/pom.xml | 5 +-
.../pipeline/core/job/api/PipelineAPIFactory.java | 29 +++-
.../PipelineContextManagerLifecycleListener.java | 16 +-
.../core/job/api/PipelineAPIFactoryTest.java | 172 +++++++++++++++++++
...ipelineContextManagerLifecycleListenerTest.java | 191 +++++++++++++++++++++
.../collect/StatisticsCollectJobWorker.java | 84 ++++-----
.../spi/CoordinatorRegistryCenterProvider.java | 43 +++++
.../collect/StatisticsCollectJobWorkerTest.java | 120 +++++++++----
.../cluster/repository/provider/zookeeper/pom.xml | 10 ++
...ZookeeperCoordinatorRegistryCenterProvider.java | 52 +++---
....schedule.spi.CoordinatorRegistryCenterProvider | 18 ++
...eeperCoordinatorRegistryCenterProviderTest.java | 131 ++++++++++++++
pom.xml | 5 +
13 files changed, 760 insertions(+), 116 deletions(-)
diff --git a/kernel/data-pipeline/core/pom.xml
b/kernel/data-pipeline/core/pom.xml
index ce1149c12dd..3c343a381e4 100644
--- a/kernel/data-pipeline/core/pom.xml
+++ b/kernel/data-pipeline/core/pom.xml
@@ -34,13 +34,12 @@
</dependency>
<dependency>
<groupId>org.apache.shardingsphere</groupId>
-
<artifactId>shardingsphere-standalone-mode-repository-memory</artifactId>
+ <artifactId>shardingsphere-schedule-core</artifactId>
<version>${project.version}</version>
</dependency>
- <!-- TODO remove cluster mode dependency after 3 modes have equivalent
features -->
<dependency>
<groupId>org.apache.shardingsphere</groupId>
-
<artifactId>shardingsphere-cluster-mode-repository-zookeeper</artifactId>
+
<artifactId>shardingsphere-standalone-mode-repository-memory</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/api/PipelineAPIFactory.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/api/PipelineAPIFactory.java
index 47d1659dee2..19b5999e52d 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/api/PipelineAPIFactory.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/api/PipelineAPIFactory.java
@@ -25,7 +25,6 @@ import org.apache.commons.lang3.concurrent.LazyInitializer;
import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
import
org.apache.shardingsphere.data.pipeline.core.context.PipelineContextManager;
import
org.apache.shardingsphere.data.pipeline.core.metadata.node.PipelineMetaDataNode;
-import
org.apache.shardingsphere.data.pipeline.core.registrycenter.elasticjob.CoordinatorRegistryCenterInitializer;
import
org.apache.shardingsphere.data.pipeline.core.registrycenter.repository.PipelineGovernanceFacade;
import
org.apache.shardingsphere.elasticjob.lite.lifecycle.api.JobConfigurationAPI;
import org.apache.shardingsphere.elasticjob.lite.lifecycle.api.JobOperateAPI;
@@ -36,9 +35,10 @@ import
org.apache.shardingsphere.elasticjob.lite.lifecycle.internal.settings.Job
import
org.apache.shardingsphere.elasticjob.lite.lifecycle.internal.statistics.JobStatisticsAPIImpl;
import
org.apache.shardingsphere.elasticjob.lite.lifecycle.internal.statistics.ShardingStatisticsAPIImpl;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
-import org.apache.shardingsphere.infra.config.mode.ModeConfiguration;
-import org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import
org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+import
org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
+import
org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -118,6 +118,20 @@ public final class PipelineAPIFactory {
return RegistryCenterHolder.getInstance(contextKey).registryCenter;
}
+ /**
+ * Close pipeline APIs.
+ *
+ * @param contextKey context key
+ */
+ public static void close(final PipelineContextKey contextKey) {
+ ElasticJobAPIHolder.INSTANCE_MAP.remove(contextKey);
+ GOVERNANCE_FACADE_MAP.remove(contextKey);
+ RegistryCenterHolder registryCenterHolder =
RegistryCenterHolder.INSTANCE_MAP.remove(contextKey);
+ if (null != registryCenterHolder) {
+ registryCenterHolder.registryCenter.close();
+ }
+ }
+
private static final class ElasticJobAPIHolder {
private static final Map<PipelineContextKey, ElasticJobAPIHolder>
INSTANCE_MAP = new ConcurrentHashMap<>();
@@ -154,12 +168,9 @@ public final class PipelineAPIFactory {
}
private CoordinatorRegistryCenter createRegistryCenter(final
PipelineContextKey contextKey) {
- CoordinatorRegistryCenterInitializer registryCenterInitializer =
new CoordinatorRegistryCenterInitializer();
- ModeConfiguration modeConfig =
PipelineContextManager.getContext(contextKey).getComputeNodeInstanceContext().getModeConfiguration();
- String elasticJobNamespace =
PipelineMetaDataNode.getElasticJobNamespace();
- String clusterType = modeConfig.getRepository().getType();
-
ShardingSpherePreconditions.checkState("ZooKeeper".equals(clusterType), () ->
new IllegalArgumentException("Unsupported cluster type: " + clusterType));
- return
registryCenterInitializer.createZookeeperRegistryCenter(modeConfig,
elasticJobNamespace);
+ ClusterPersistRepositoryConfiguration repositoryConfig =
(ClusterPersistRepositoryConfiguration)
PipelineContextManager.getContext(contextKey)
+
.getComputeNodeInstanceContext().getModeConfiguration().getRepository();
+ return
TypedSPILoader.getService(CoordinatorRegistryCenterProvider.class,
repositoryConfig.getType()).create(repositoryConfig,
PipelineMetaDataNode.getElasticJobNamespace());
}
public static RegistryCenterHolder getInstance(final
PipelineContextKey contextKey) {
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/listener/PipelineContextManagerLifecycleListener.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/listener/PipelineContextManagerLifecycleListener.java
index 52c7963b453..0a40a189441 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/listener/PipelineContextManagerLifecycleListener.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/listener/PipelineContextManagerLifecycleListener.java
@@ -32,9 +32,11 @@ import
org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
import org.apache.shardingsphere.elasticjob.infra.spi.ElasticJobServiceLoader;
import
org.apache.shardingsphere.elasticjob.lite.lifecycle.api.JobConfigurationAPI;
import org.apache.shardingsphere.elasticjob.lite.lifecycle.domain.JobBriefInfo;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.mode.manager.ContextManager;
import
org.apache.shardingsphere.mode.manager.listener.ContextManagerLifecycleListener;
import
org.apache.shardingsphere.mode.manager.listener.ContextManagerLifecycleListenerModeRequired;
+import
org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider;
import java.util.List;
import java.util.stream.Collectors;
@@ -54,6 +56,11 @@ public final class PipelineContextManagerLifecycleListener
implements ContextMan
}
PipelineContextKey contextKey = new
PipelineContextKey(preSelectedDatabaseName,
contextManager.getComputeNodeInstanceContext().getInstance().getMetaData().getType());
PipelineContextManager.putContext(contextKey, contextManager);
+ String repositoryType =
contextManager.getComputeNodeInstanceContext().getModeConfiguration().getRepository().getType();
+ if
(!TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
repositoryType).isPresent()) {
+ log.warn("Can not resume pipeline jobs because coordinator
registry center provider is unavailable for repository type: {}",
repositoryType);
+ return;
+ }
PipelineMetaDataNodeWatcher.init(contextKey);
ElasticJobServiceLoader.registerTypedService(ElasticJobListener.class);
try {
@@ -98,7 +105,12 @@ public final class PipelineContextManagerLifecycleListener
implements ContextMan
@Override
public void onDestroyed(final ContextManager contextManager) {
- PipelineContextManager.removeContext(
- new
PipelineContextKey(contextManager.getPreSelectedDatabaseName(),
contextManager.getComputeNodeInstanceContext().getInstance().getMetaData().getType()));
+ PipelineContextKey contextKey = new PipelineContextKey(
+ contextManager.getPreSelectedDatabaseName(),
contextManager.getComputeNodeInstanceContext().getInstance().getMetaData().getType());
+ try {
+ PipelineAPIFactory.close(contextKey);
+ } finally {
+ PipelineContextManager.removeContext(contextKey);
+ }
}
}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/api/PipelineAPIFactoryTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/api/PipelineAPIFactoryTest.java
new file mode 100644
index 00000000000..bdcda1c865c
--- /dev/null
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/api/PipelineAPIFactoryTest.java
@@ -0,0 +1,172 @@
+/*
+ * 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.data.pipeline.core.job.api;
+
+import lombok.SneakyThrows;
+import org.apache.commons.lang3.concurrent.ConcurrentException;
+import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
+import
org.apache.shardingsphere.data.pipeline.core.context.PipelineContextManager;
+import
org.apache.shardingsphere.data.pipeline.core.metadata.node.PipelineMetaDataNode;
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+import org.apache.shardingsphere.infra.config.mode.ModeConfiguration;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
+import
org.apache.shardingsphere.infra.spi.exception.ServiceProviderNotFoundException;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.mode.manager.ContextManager;
+import
org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+import
org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
+import
org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.internal.configuration.plugins.Plugins;
+
+import java.util.Map;
+import java.util.Properties;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PipelineAPIFactoryTest {
+
+ @Test
+ void assertGetRegistryCenterViaProvider() {
+ PipelineContextKey contextKey = new PipelineContextKey("foo_db",
InstanceType.JDBC);
+ CoordinatorRegistryCenter registryCenter =
mock(CoordinatorRegistryCenter.class);
+ CoordinatorRegistryCenterProvider provider =
mock(CoordinatorRegistryCenterProvider.class);
+ try (
+ MockedStatic<PipelineContextManager> contextManager =
mockStatic(PipelineContextManager.class);
+ MockedStatic<TypedSPILoader> serviceLoader =
mockStatic(TypedSPILoader.class)) {
+ ClusterPersistRepositoryConfiguration repositoryConfig =
mockContext(contextKey, contextManager);
+ serviceLoader.when(() ->
TypedSPILoader.getService(CoordinatorRegistryCenterProvider.class,
repositoryConfig.getType())).thenReturn(provider);
+ when(provider.create(repositoryConfig,
PipelineMetaDataNode.getElasticJobNamespace())).thenReturn(registryCenter);
+ assertThat(PipelineAPIFactory.getRegistryCenter(contextKey),
sameInstance(registryCenter));
+ assertThat(PipelineAPIFactory.getRegistryCenter(contextKey),
sameInstance(registryCenter));
+ verify(provider).create(repositoryConfig,
PipelineMetaDataNode.getElasticJobNamespace());
+ } finally {
+ PipelineAPIFactory.close(contextKey);
+ }
+ verify(registryCenter).close();
+ }
+
+ @Test
+ void assertClose() {
+ PipelineContextKey contextKey = new PipelineContextKey("bar_db",
InstanceType.JDBC);
+ CoordinatorRegistryCenter registryCenter =
mock(CoordinatorRegistryCenter.class);
+ CoordinatorRegistryCenterProvider provider =
mock(CoordinatorRegistryCenterProvider.class);
+ try (
+ MockedStatic<PipelineContextManager> contextManager =
mockStatic(PipelineContextManager.class);
+ MockedStatic<TypedSPILoader> serviceLoader =
mockStatic(TypedSPILoader.class)) {
+ ClusterPersistRepositoryConfiguration repositoryConfig =
mockContext(contextKey, contextManager);
+ serviceLoader.when(() ->
TypedSPILoader.getService(CoordinatorRegistryCenterProvider.class,
repositoryConfig.getType())).thenReturn(provider);
+ when(provider.create(repositoryConfig,
PipelineMetaDataNode.getElasticJobNamespace())).thenReturn(registryCenter);
+ PipelineAPIFactory.getJobStatisticsAPI(contextKey);
+ PipelineAPIFactory.getJobConfigurationAPI(contextKey);
+ PipelineAPIFactory.getJobOperateAPI(contextKey);
+ PipelineAPIFactory.getShardingStatisticsAPI(contextKey);
+ PipelineAPIFactory.getPipelineGovernanceFacade(contextKey);
+ assertTrue(getMap(PipelineAPIFactory.class,
"GOVERNANCE_FACADE_MAP").containsKey(contextKey));
+ assertTrue(getMap(getHolderClass("ElasticJobAPIHolder"),
"INSTANCE_MAP").containsKey(contextKey));
+ assertTrue(getMap(getHolderClass("RegistryCenterHolder"),
"INSTANCE_MAP").containsKey(contextKey));
+ PipelineAPIFactory.close(contextKey);
+ assertFalse(getMap(PipelineAPIFactory.class,
"GOVERNANCE_FACADE_MAP").containsKey(contextKey));
+ assertFalse(getMap(getHolderClass("ElasticJobAPIHolder"),
"INSTANCE_MAP").containsKey(contextKey));
+ assertFalse(getMap(getHolderClass("RegistryCenterHolder"),
"INSTANCE_MAP").containsKey(contextKey));
+ PipelineAPIFactory.close(contextKey);
+ }
+ verify(registryCenter, times(1)).close();
+ }
+
+ @Test
+ void assertCloseWithoutCreatingResources() {
+ PipelineContextKey contextKey = new PipelineContextKey("empty_db",
InstanceType.JDBC);
+ try (MockedStatic<TypedSPILoader> serviceLoader =
mockStatic(TypedSPILoader.class)) {
+ PipelineAPIFactory.close(contextKey);
+ serviceLoader.verifyNoInteractions();
+ }
+ assertFalse(getMap(PipelineAPIFactory.class,
"GOVERNANCE_FACADE_MAP").containsKey(contextKey));
+ assertFalse(getMap(getHolderClass("ElasticJobAPIHolder"),
"INSTANCE_MAP").containsKey(contextKey));
+ assertFalse(getMap(getHolderClass("RegistryCenterHolder"),
"INSTANCE_MAP").containsKey(contextKey));
+ }
+
+ @Test
+ void assertGetRegistryCenterWithUnsupportedProvider() {
+ PipelineContextKey contextKey = new
PipelineContextKey("unsupported_db", InstanceType.JDBC);
+ try (MockedStatic<PipelineContextManager> contextManager =
mockStatic(PipelineContextManager.class)) {
+ ClusterPersistRepositoryConfiguration repositoryConfig = new
ClusterPersistRepositoryConfiguration("NOT_REGISTERED", "namespace",
"serverLists", new Properties());
+ ContextManager contextManagerInstance = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ contextManager.when(() ->
PipelineContextManager.getContext(contextKey)).thenReturn(contextManagerInstance);
+ whenModeConfiguration(contextManagerInstance, repositoryConfig);
+ ServiceProviderNotFoundException actual =
assertThrows(ServiceProviderNotFoundException.class, () ->
PipelineAPIFactory.getRegistryCenter(contextKey));
+ assertThat(actual.getMessage(), containsString("NOT_REGISTERED"));
+ } finally {
+ PipelineAPIFactory.close(contextKey);
+ }
+ }
+
+ @Test
+ void assertGetPipelineGovernanceFacadePropagatesInitializationFailure() {
+ PipelineContextKey contextKey = new PipelineContextKey("error_db",
InstanceType.JDBC);
+ ConcurrentException expected = new ConcurrentException(new
Exception("expected"));
+ try (MockedStatic<PipelineContextManager> contextManager =
mockStatic(PipelineContextManager.class)) {
+ contextManager.when(() ->
PipelineContextManager.getContext(contextKey)).thenAnswer(invocation -> {
+ throw expected;
+ });
+ assertThat(assertThrows(ConcurrentException.class, () ->
PipelineAPIFactory.getPipelineGovernanceFacade(contextKey)),
sameInstance(expected));
+ } finally {
+ PipelineAPIFactory.close(contextKey);
+ }
+ }
+
+ private ClusterPersistRepositoryConfiguration mockContext(final
PipelineContextKey contextKey, final MockedStatic<PipelineContextManager>
pipelineContextManager) {
+ ClusterPersistRepositoryConfiguration result = new
ClusterPersistRepositoryConfiguration("FIXTURE", "namespace", "serverLists",
new Properties());
+ ContextManager contextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ pipelineContextManager.when(() ->
PipelineContextManager.getContext(contextKey)).thenReturn(contextManager);
+ whenModeConfiguration(contextManager, result);
+ whenRepository(contextManager);
+ return result;
+ }
+
+ private void whenModeConfiguration(final ContextManager contextManager,
final ClusterPersistRepositoryConfiguration repositoryConfig) {
+
when(contextManager.getComputeNodeInstanceContext().getModeConfiguration()).thenReturn(new
ModeConfiguration("Cluster", repositoryConfig));
+ }
+
+ private void whenRepository(final ContextManager contextManager) {
+
when(contextManager.getPersistServiceFacade().getRepository()).thenReturn(mock(ClusterPersistRepository.class));
+ }
+
+ @SneakyThrows(ClassNotFoundException.class)
+ private Class<?> getHolderClass(final String simpleName) {
+ return Class.forName(PipelineAPIFactory.class.getName() + "$" +
simpleName);
+ }
+
+ @SuppressWarnings("unchecked")
+ @SneakyThrows(ReflectiveOperationException.class)
+ private Map<PipelineContextKey, ?> getMap(final Class<?> owner, final
String fieldName) {
+ return (Map<PipelineContextKey, ?>)
Plugins.getMemberAccessor().get(owner.getDeclaredField(fieldName), owner);
+ }
+}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/listener/PipelineContextManagerLifecycleListenerTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/listener/PipelineContextManagerLifecycleListenerTest.java
new file mode 100644
index 00000000000..7cf492c470b
--- /dev/null
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/listener/PipelineContextManagerLifecycleListenerTest.java
@@ -0,0 +1,191 @@
+/*
+ * 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.data.pipeline.core.listener;
+
+import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
+import
org.apache.shardingsphere.data.pipeline.core.context.PipelineContextManager;
+import
org.apache.shardingsphere.data.pipeline.core.exception.job.PipelineJobNotFoundException;
+import org.apache.shardingsphere.data.pipeline.core.job.api.PipelineAPIFactory;
+import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobIdUtils;
+import
org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobManager;
+import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobType;
+import
org.apache.shardingsphere.data.pipeline.core.metadata.node.PipelineMetaDataNodeWatcher;
+import org.apache.shardingsphere.database.connector.core.DefaultDatabase;
+import org.apache.shardingsphere.elasticjob.infra.listener.ElasticJobListener;
+import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
+import org.apache.shardingsphere.elasticjob.infra.spi.ElasticJobServiceLoader;
+import
org.apache.shardingsphere.elasticjob.lite.lifecycle.api.JobConfigurationAPI;
+import
org.apache.shardingsphere.elasticjob.lite.lifecycle.api.JobStatisticsAPI;
+import org.apache.shardingsphere.elasticjob.lite.lifecycle.domain.JobBriefInfo;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.mode.manager.ContextManager;
+import
org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Optional;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PipelineContextManagerLifecycleListenerTest {
+
+ @Test
+ void assertInitializeWithDefaultDatabase() {
+ ContextManager contextManager = mock(ContextManager.class);
+
when(contextManager.getPreSelectedDatabaseName()).thenReturn(DefaultDatabase.LOGIC_NAME);
+ new
PipelineContextManagerLifecycleListener().onInitialized(contextManager);
+ }
+
+ @Test
+ void assertInitializeAndResumeEnabledJobs() {
+ JobConfigurationAPI jobConfigAPI = mock(JobConfigurationAPI.class);
+ JobStatisticsAPI jobStatisticsAPI = mock(JobStatisticsAPI.class);
+
when(jobStatisticsAPI.getAllJobsBriefInfo()).thenReturn(Arrays.asList(createJobBriefInfo("_internal"),
createJobBriefInfo("invalid"),
+ createJobBriefInfo("consistency"),
createJobBriefInfo("missing"), createJobBriefInfo("disabled"),
createJobBriefInfo("enabled")));
+ PipelineJobType<?> consistencyCheckJobType =
mock(PipelineJobType.class);
+
when(consistencyCheckJobType.getType()).thenReturn("CONSISTENCY_CHECK");
+ PipelineJobType<?> jobType = mock(PipelineJobType.class);
+ when(jobType.getType()).thenReturn("FIXTURE");
+ when(jobConfigAPI.getJobConfiguration("missing")).thenThrow(new
PipelineJobNotFoundException("missing"));
+ JobConfigurationPOJO disabledJobConfig =
mock(JobConfigurationPOJO.class);
+ when(disabledJobConfig.isDisabled()).thenReturn(true);
+
when(jobConfigAPI.getJobConfiguration("disabled")).thenReturn(disabledJobConfig);
+
when(jobConfigAPI.getJobConfiguration("enabled")).thenReturn(mock(JobConfigurationPOJO.class));
+ ContextManager contextManager = mockContextManager();
+ PipelineContextKey contextKey = new PipelineContextKey("foo_db",
InstanceType.JDBC);
+ try (
+ MockedStatic<PipelineContextManager> pipelineContextManager =
mockStatic(PipelineContextManager.class);
+ MockedStatic<PipelineMetaDataNodeWatcher> watcher =
mockStatic(PipelineMetaDataNodeWatcher.class);
+ MockedStatic<ElasticJobServiceLoader> serviceLoader =
mockStatic(ElasticJobServiceLoader.class);
+ MockedStatic<TypedSPILoader> typedSPILoader =
mockStatic(TypedSPILoader.class);
+ MockedStatic<PipelineAPIFactory> pipelineAPIFactory =
mockStatic(PipelineAPIFactory.class);
+ MockedStatic<PipelineJobIdUtils> jobIdUtils =
mockStatic(PipelineJobIdUtils.class);
+ MockedConstruction<PipelineJobManager> jobManagerConstruction
= mockConstruction(PipelineJobManager.class)) {
+ typedSPILoader.when(() ->
TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"FIXTURE")).thenReturn(Optional.of(mock(CoordinatorRegistryCenterProvider.class)));
+ pipelineAPIFactory.when(() ->
PipelineAPIFactory.getJobConfigurationAPI(contextKey)).thenReturn(jobConfigAPI);
+ pipelineAPIFactory.when(() ->
PipelineAPIFactory.getJobStatisticsAPI(contextKey)).thenReturn(jobStatisticsAPI);
+ jobIdUtils.when(() ->
PipelineJobIdUtils.parseJobType("invalid")).thenThrow(new
IllegalArgumentException("expected"));
+ jobIdUtils.when(() ->
PipelineJobIdUtils.parseJobType("consistency")).thenReturn(consistencyCheckJobType);
+ jobIdUtils.when(() ->
PipelineJobIdUtils.parseJobType("missing")).thenReturn(jobType);
+ jobIdUtils.when(() ->
PipelineJobIdUtils.parseJobType("disabled")).thenReturn(jobType);
+ jobIdUtils.when(() ->
PipelineJobIdUtils.parseJobType("enabled")).thenReturn(jobType);
+ new
PipelineContextManagerLifecycleListener().onInitialized(contextManager);
+ pipelineContextManager.verify(() ->
PipelineContextManager.putContext(contextKey, contextManager));
+ watcher.verify(() -> PipelineMetaDataNodeWatcher.init(contextKey));
+ serviceLoader.verify(() ->
ElasticJobServiceLoader.registerTypedService(ElasticJobListener.class));
+ assertThat(jobManagerConstruction.constructed().size(), is(1));
+
verify(jobManagerConstruction.constructed().get(0)).resume("enabled");
+ }
+ }
+
+ @Test
+ void assertInitializeWithoutCoordinatorRegistryCenterProvider() {
+ ContextManager contextManager = mockContextManager();
+ PipelineContextKey contextKey = new PipelineContextKey("foo_db",
InstanceType.JDBC);
+ try (
+ MockedStatic<PipelineContextManager> pipelineContextManager =
mockStatic(PipelineContextManager.class);
+ MockedStatic<PipelineMetaDataNodeWatcher> watcher =
mockStatic(PipelineMetaDataNodeWatcher.class);
+ MockedStatic<ElasticJobServiceLoader> serviceLoader =
mockStatic(ElasticJobServiceLoader.class);
+ MockedStatic<TypedSPILoader> typedSPILoader =
mockStatic(TypedSPILoader.class);
+ MockedStatic<PipelineAPIFactory> pipelineAPIFactory =
mockStatic(PipelineAPIFactory.class)) {
+ typedSPILoader.when(() ->
TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"FIXTURE")).thenReturn(Optional.empty());
+ new
PipelineContextManagerLifecycleListener().onInitialized(contextManager);
+ pipelineContextManager.verify(() ->
PipelineContextManager.putContext(contextKey, contextManager));
+ watcher.verifyNoInteractions();
+ serviceLoader.verifyNoInteractions();
+ pipelineAPIFactory.verifyNoInteractions();
+ }
+ }
+
+ @Test
+ void assertInitializeIgnoresDispatchFailure() {
+ ContextManager contextManager = mockContextManager();
+ PipelineContextKey contextKey = new PipelineContextKey("foo_db",
InstanceType.JDBC);
+ try (
+ MockedStatic<PipelineContextManager> ignoredContextManager =
mockStatic(PipelineContextManager.class);
+ MockedStatic<PipelineMetaDataNodeWatcher> ignoredWatcher =
mockStatic(PipelineMetaDataNodeWatcher.class);
+ MockedStatic<ElasticJobServiceLoader> ignoredServiceLoader =
mockStatic(ElasticJobServiceLoader.class);
+ MockedStatic<TypedSPILoader> typedSPILoader =
mockStatic(TypedSPILoader.class);
+ MockedStatic<PipelineAPIFactory> pipelineAPIFactory =
mockStatic(PipelineAPIFactory.class)) {
+ typedSPILoader.when(() ->
TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"FIXTURE")).thenReturn(Optional.of(mock(CoordinatorRegistryCenterProvider.class)));
+ pipelineAPIFactory.when(() ->
PipelineAPIFactory.getJobConfigurationAPI(contextKey)).thenThrow(new
RuntimeException("expected"));
+ assertDoesNotThrow(() -> new
PipelineContextManagerLifecycleListener().onInitialized(contextManager));
+ }
+ }
+
+ @Test
+ void assertDestroyReleasesPipelineResourcesBeforeRemovingContext() {
+ ContextManager contextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ when(contextManager.getPreSelectedDatabaseName()).thenReturn("foo_db");
+
when(contextManager.getComputeNodeInstanceContext().getInstance().getMetaData().getType()).thenReturn(InstanceType.JDBC);
+ PipelineContextKey contextKey = new PipelineContextKey("foo_db",
InstanceType.JDBC);
+ List<String> invocations = new LinkedList<>();
+ try (
+ MockedStatic<PipelineAPIFactory> pipelineAPIFactory =
mockStatic(PipelineAPIFactory.class);
+ MockedStatic<PipelineContextManager> pipelineContextManager =
mockStatic(PipelineContextManager.class)) {
+ pipelineAPIFactory.when(() ->
PipelineAPIFactory.close(contextKey)).thenAnswer(invocation ->
invocations.add("close"));
+ pipelineContextManager.when(() ->
PipelineContextManager.removeContext(contextKey)).thenAnswer(invocation ->
invocations.add("remove"));
+ new
PipelineContextManagerLifecycleListener().onDestroyed(contextManager);
+ }
+ assertThat(invocations, contains("close", "remove"));
+ }
+
+ @Test
+ void assertDestroyRemovesContextWhenResourceReleaseFailed() {
+ ContextManager contextManager = mockContextManager();
+ PipelineContextKey contextKey = new PipelineContextKey("foo_db",
InstanceType.JDBC);
+ RuntimeException expected = new RuntimeException("expected");
+ try (
+ MockedStatic<PipelineAPIFactory> pipelineAPIFactory =
mockStatic(PipelineAPIFactory.class);
+ MockedStatic<PipelineContextManager> pipelineContextManager =
mockStatic(PipelineContextManager.class)) {
+ pipelineAPIFactory.when(() ->
PipelineAPIFactory.close(contextKey)).thenThrow(expected);
+ assertThat(assertThrows(RuntimeException.class, () -> new
PipelineContextManagerLifecycleListener().onDestroyed(contextManager)),
sameInstance(expected));
+ pipelineContextManager.verify(() ->
PipelineContextManager.removeContext(contextKey));
+ }
+ }
+
+ private ContextManager mockContextManager() {
+ ContextManager result = mock(ContextManager.class, RETURNS_DEEP_STUBS);
+ when(result.getPreSelectedDatabaseName()).thenReturn("foo_db");
+
when(result.getComputeNodeInstanceContext().getInstance().getMetaData().getType()).thenReturn(InstanceType.JDBC);
+
when(result.getComputeNodeInstanceContext().getModeConfiguration().getRepository().getType()).thenReturn("FIXTURE");
+ return result;
+ }
+
+ private JobBriefInfo createJobBriefInfo(final String jobName) {
+ JobBriefInfo result = new JobBriefInfo();
+ result.setJobName(jobName);
+ return result;
+ }
+}
diff --git
a/kernel/schedule/core/src/main/java/org/apache/shardingsphere/schedule/core/job/statistics/collect/StatisticsCollectJobWorker.java
b/kernel/schedule/core/src/main/java/org/apache/shardingsphere/schedule/core/job/statistics/collect/StatisticsCollectJobWorker.java
index c781e1630b0..efd7ed9472a 100644
---
a/kernel/schedule/core/src/main/java/org/apache/shardingsphere/schedule/core/job/statistics/collect/StatisticsCollectJobWorker.java
+++
b/kernel/schedule/core/src/main/java/org/apache/shardingsphere/schedule/core/job/statistics/collect/StatisticsCollectJobWorker.java
@@ -25,18 +25,17 @@ import
org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.ScheduleJobB
import
org.apache.shardingsphere.elasticjob.lite.lifecycle.internal.operate.JobOperateAPIImpl;
import
org.apache.shardingsphere.elasticjob.lite.lifecycle.internal.settings.JobConfigurationAPIImpl;
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.infra.config.mode.ModeConfiguration;
import
org.apache.shardingsphere.infra.config.props.temporary.TemporaryConfigurationPropertyKey;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.mode.manager.ContextManager;
import
org.apache.shardingsphere.mode.node.path.engine.generator.NodePathGenerator;
import
org.apache.shardingsphere.mode.node.path.type.database.statistics.StatisticsJobNodePath;
import
org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
+import
org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider;
import org.quartz.CronExpression;
import java.util.Optional;
-import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -63,48 +62,35 @@ public final class StatisticsCollectJobWorker {
*/
public void initialize(final ContextManager contextManager) {
if (WORKER_INITIALIZED.compareAndSet(false, true)) {
- ModeConfiguration modeConfig =
contextManager.getComputeNodeInstanceContext().getModeConfiguration();
- if (!"ZooKeeper".equals(modeConfig.getRepository().getType())) {
- log.warn("Can not collect statistics because of unsupported
cluster type: {}", modeConfig.getRepository().getType());
- return;
+ try {
+ ModeConfiguration modeConfig =
contextManager.getComputeNodeInstanceContext().getModeConfiguration();
+ Optional<CoordinatorRegistryCenterProvider> provider =
TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
modeConfig.getRepository().getType());
+ if (!provider.isPresent()) {
+ log.warn("Can not collect statistics because of
unsupported cluster type: {}", modeConfig.getRepository().getType());
+ WORKER_INITIALIZED.set(false);
+ return;
+ }
+ StatisticsCollectJobWorker.contextManager = contextManager;
+ ClusterPersistRepositoryConfiguration repositoryConfig =
(ClusterPersistRepositoryConfiguration) modeConfig.getRepository();
+ registryCenter = provider.get().create(repositoryConfig,
NodePathGenerator.toPath(new StatisticsJobNodePath()));
+ scheduleJobBootstrap = new
ScheduleJobBootstrap(registryCenter, new StatisticsCollectJob(contextManager),
createJobConfiguration());
+ scheduleJobBootstrap.schedule();
+ new JobOperateAPIImpl(registryCenter).trigger(JOB_NAME);
+ // CHECKSTYLE:OFF
+ } catch (final RuntimeException ex) {
+ // CHECKSTYLE:ON
+ try {
+ destroy();
+ // CHECKSTYLE:OFF
+ } catch (final RuntimeException cleanupEx) {
+ // CHECKSTYLE:ON
+ ex.addSuppressed(cleanupEx);
+ }
+ throw ex;
}
- StatisticsCollectJobWorker.contextManager = contextManager;
- registryCenter = createRegistryCenter(modeConfig);
- scheduleJobBootstrap = new ScheduleJobBootstrap(registryCenter,
new StatisticsCollectJob(contextManager), createJobConfiguration());
- scheduleJobBootstrap.schedule();
- new JobOperateAPIImpl(registryCenter).trigger(JOB_NAME);
}
}
- private CoordinatorRegistryCenter createRegistryCenter(final
ModeConfiguration modeConfig) {
- ClusterPersistRepositoryConfiguration repositoryConfig =
(ClusterPersistRepositoryConfiguration) modeConfig.getRepository();
- String namespace = repositoryConfig.getNamespace() +
NodePathGenerator.toPath(new StatisticsJobNodePath());
- CoordinatorRegistryCenter result = new
ZookeeperRegistryCenter(getZookeeperConfiguration(repositoryConfig, namespace));
- result.init();
- return result;
- }
-
- private ZookeeperConfiguration getZookeeperConfiguration(final
ClusterPersistRepositoryConfiguration repositoryConfig, final String namespace)
{
- // TODO Merge registry center code in ElasticJob and ShardingSphere
mode; Use SPI to load impl
- ZookeeperConfiguration result = new
ZookeeperConfiguration(repositoryConfig.getServerLists(), namespace);
- Properties props = repositoryConfig.getProps();
- int retryIntervalMilliseconds =
props.containsKey("retryIntervalMilliseconds") ?
Integer.parseInt(props.get("retryIntervalMilliseconds").toString()) : 500;
- int maxRetries = props.containsKey("maxRetries") ?
Integer.parseInt(props.get("maxRetries").toString()) : 3;
- result.setBaseSleepTimeMilliseconds(retryIntervalMilliseconds);
- result.setMaxRetries(maxRetries);
- result.setMaxSleepTimeMilliseconds(retryIntervalMilliseconds *
maxRetries);
- int timeToLiveSeconds = props.containsKey("timeToLiveSeconds") ?
Integer.parseInt(props.get("timeToLiveSeconds").toString()) : 60;
- if (0 != timeToLiveSeconds) {
- result.setSessionTimeoutMilliseconds(timeToLiveSeconds * 1000);
- }
- int operationTimeoutMilliseconds =
props.containsKey("operationTimeoutMilliseconds") ?
Integer.parseInt(props.get("operationTimeoutMilliseconds").toString()) : 500;
- if (0 != operationTimeoutMilliseconds) {
-
result.setConnectionTimeoutMilliseconds(operationTimeoutMilliseconds);
- }
- result.setDigest(props.getProperty("digest"));
- return result;
- }
-
private JobConfiguration createJobConfiguration() {
String jobCron =
contextManager.getMetaDataContexts().getMetaData().getTemporaryProps().getValue(TemporaryConfigurationPropertyKey.PROXY_META_DATA_COLLECTOR_CRON);
if (!CronExpression.isValidExpression(jobCron)) {
@@ -139,11 +125,17 @@ public final class StatisticsCollectJobWorker {
*/
public void destroy() {
if (WORKER_INITIALIZED.compareAndSet(true, false)) {
-
Optional.ofNullable(scheduleJobBootstrap).ifPresent(ScheduleJobBootstrap::shutdown);
- scheduleJobBootstrap = null;
-
Optional.ofNullable(registryCenter).ifPresent(CoordinatorRegistryCenter::close);
- registryCenter = null;
- contextManager = null;
+ try {
+
Optional.ofNullable(scheduleJobBootstrap).ifPresent(ScheduleJobBootstrap::shutdown);
+ } finally {
+ scheduleJobBootstrap = null;
+ try {
+
Optional.ofNullable(registryCenter).ifPresent(CoordinatorRegistryCenter::close);
+ } finally {
+ registryCenter = null;
+ contextManager = null;
+ }
+ }
}
}
}
diff --git
a/kernel/schedule/core/src/main/java/org/apache/shardingsphere/schedule/spi/CoordinatorRegistryCenterProvider.java
b/kernel/schedule/core/src/main/java/org/apache/shardingsphere/schedule/spi/CoordinatorRegistryCenterProvider.java
new file mode 100644
index 00000000000..51b6dea4b1e
--- /dev/null
+++
b/kernel/schedule/core/src/main/java/org/apache/shardingsphere/schedule/spi/CoordinatorRegistryCenterProvider.java
@@ -0,0 +1,43 @@
+/*
+ * 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.schedule.spi;
+
+import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPI;
+import
org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
+
+/**
+ * Coordinator registry center provider.
+ *
+ * <p>Implementations must not retain invocation state or create resources in
their constructor or SPI initialization method.</p>
+ */
+@SingletonSPI
+public interface CoordinatorRegistryCenterProvider extends TypedSPI {
+
+ /**
+ * Create an initialized coordinator registry center.
+ *
+ * <p>The caller owns the returned registry center and must close it. If
initialization fails, the provider must close any partially created registry
center before propagating the failure.</p>
+ *
+ * @param repositoryConfig cluster persist repository configuration
+ * @param namespaceRelativePath namespace relative path; {@code null}
means that no relative path is appended
+ * @return initialized coordinator registry center
+ */
+ CoordinatorRegistryCenter create(ClusterPersistRepositoryConfiguration
repositoryConfig, String namespaceRelativePath);
+}
diff --git
a/kernel/schedule/core/src/test/java/org/apache/shardingsphere/schedule/core/job/statistics/collect/StatisticsCollectJobWorkerTest.java
b/kernel/schedule/core/src/test/java/org/apache/shardingsphere/schedule/core/job/statistics/collect/StatisticsCollectJobWorkerTest.java
index b489fcf5f03..b3b775f2c0a 100644
---
a/kernel/schedule/core/src/test/java/org/apache/shardingsphere/schedule/core/job/statistics/collect/StatisticsCollectJobWorkerTest.java
+++
b/kernel/schedule/core/src/test/java/org/apache/shardingsphere/schedule/core/job/statistics/collect/StatisticsCollectJobWorkerTest.java
@@ -24,14 +24,17 @@ import
org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.ScheduleJobB
import
org.apache.shardingsphere.elasticjob.lite.lifecycle.internal.operate.JobOperateAPIImpl;
import
org.apache.shardingsphere.elasticjob.lite.lifecycle.internal.settings.JobConfigurationAPIImpl;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
-import
org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;
import org.apache.shardingsphere.infra.config.mode.ModeConfiguration;
import
org.apache.shardingsphere.infra.config.props.temporary.TemporaryConfigurationProperties;
import
org.apache.shardingsphere.infra.config.props.temporary.TemporaryConfigurationPropertyKey;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.util.props.PropertiesBuilder;
import org.apache.shardingsphere.infra.util.props.PropertiesBuilder.Property;
import org.apache.shardingsphere.mode.manager.ContextManager;
+import
org.apache.shardingsphere.mode.node.path.engine.generator.NodePathGenerator;
+import
org.apache.shardingsphere.mode.node.path.type.database.statistics.StatisticsJobNodePath;
import
org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
+import
org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -40,21 +43,30 @@ import org.mockito.Answers;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
import org.mockito.internal.configuration.plugins.Plugins;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.arrayContaining;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -62,18 +74,27 @@ import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class StatisticsCollectJobWorkerTest {
+ private static final String STATISTICS_JOB_PATH =
NodePathGenerator.toPath(new StatisticsJobNodePath());
+
private static AtomicBoolean workerInitialized = new AtomicBoolean(false);
private StatisticsCollectJobWorker jobWorker;
+ private MockedStatic<TypedSPILoader> serviceLoader;
+
@Mock(answer = Answers.RETURNS_DEEP_STUBS)
private ContextManager contextManager;
+ @Mock
+ private CoordinatorRegistryCenterProvider provider;
+
@BeforeEach
@SneakyThrows(ReflectiveOperationException.class)
void setUp() {
workerInitialized = (AtomicBoolean)
Plugins.getMemberAccessor().get(StatisticsCollectJobWorker.class.getDeclaredField("WORKER_INITIALIZED"),
StatisticsCollectJobWorker.class);
+ workerInitialized.set(false);
jobWorker = new StatisticsCollectJobWorker();
+ serviceLoader = mockStatic(TypedSPILoader.class);
}
@AfterEach
@@ -82,6 +103,7 @@ class StatisticsCollectJobWorkerTest {
setStaticField("scheduleJobBootstrap", null);
setStaticField("contextManager", null);
setStaticField("registryCenter", null);
+ serviceLoader.close();
}
@SneakyThrows(ReflectiveOperationException.class)
@@ -91,66 +113,98 @@ class StatisticsCollectJobWorkerTest {
@Test
void assertInitializeTwice() {
- jobWorker.initialize(contextManager);
- jobWorker.initialize(contextManager);
-
verify(contextManager.getComputeNodeInstanceContext()).getModeConfiguration();
- }
-
- @Test
- void assertInitializeWithZooKeeperRepository() {
- Properties repositoryProps = PropertiesBuilder.build(new
Property("retryIntervalMilliseconds", 200),
- new Property("maxRetries", 4), new
Property("timeToLiveSeconds", 5), new Property("operationTimeoutMilliseconds",
800), new Property("digest", "digest"));
-
when(contextManager.getComputeNodeInstanceContext().getModeConfiguration()).thenReturn(
- new ModeConfiguration("Cluster", new
ClusterPersistRepositoryConfiguration("ZooKeeper", "namespace",
"127.0.0.1:2181", repositoryProps)));
+ ClusterPersistRepositoryConfiguration repositoryConfig =
mockRepositoryConfiguration("FIXTURE");
+ CoordinatorRegistryCenter registryCenter =
mock(CoordinatorRegistryCenter.class);
+ serviceLoader.when(() ->
TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"FIXTURE")).thenReturn(Optional.of(provider));
+ when(provider.create(repositoryConfig,
STATISTICS_JOB_PATH)).thenReturn(registryCenter);
when(contextManager.getMetaDataContexts().getMetaData().getTemporaryProps()).thenReturn(
new
TemporaryConfigurationProperties(PropertiesBuilder.build(new
Property(TemporaryConfigurationPropertyKey.PROXY_META_DATA_COLLECTOR_CRON.getKey(),
"0 0/5 * * * ?"))));
AtomicReference<JobConfiguration> jobConfigRef = new
AtomicReference<>();
try (
- MockedConstruction<ZookeeperRegistryCenter>
registryCenterConstruction = mockConstruction(ZookeeperRegistryCenter.class);
MockedConstruction<ScheduleJobBootstrap>
scheduleJobBootstrapConstruction = mockConstruction(ScheduleJobBootstrap.class,
(mock, context) -> jobConfigRef.set((JobConfiguration)
context.arguments().get(2)));
MockedConstruction<JobOperateAPIImpl>
jobOperateAPIConstruction = mockConstruction(JobOperateAPIImpl.class)) {
jobWorker.initialize(contextManager);
- verify(registryCenterConstruction.constructed().get(0)).init();
+ jobWorker.initialize(contextManager);
+ verify(provider).create(repositoryConfig, STATISTICS_JOB_PATH);
verify(scheduleJobBootstrapConstruction.constructed().get(0)).schedule();
assertThat(jobConfigRef.get().getCron(), is("0 0/5 * * * ?"));
verify(jobOperateAPIConstruction.constructed().get(0)).trigger("statistics-collect");
+ jobWorker.destroy();
+
verify(scheduleJobBootstrapConstruction.constructed().get(0)).shutdown();
+ verify(registryCenter).close();
}
}
@Test
- void assertInitializeWithZooKeeperRepositoryUsingDefaultConfiguration() {
- Properties repositoryProps = PropertiesBuilder.build(new
Property("timeToLiveSeconds", 0), new Property("operationTimeoutMilliseconds",
0));
-
when(contextManager.getComputeNodeInstanceContext().getModeConfiguration()).thenReturn(
- new ModeConfiguration("Cluster", new
ClusterPersistRepositoryConfiguration("ZooKeeper", "namespace",
"127.0.0.1:2181", repositoryProps)));
+ void assertInitializeWithUnsupportedProvider() {
+ mockRepositoryConfiguration("UNSUPPORTED");
+ serviceLoader.when(() ->
TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"UNSUPPORTED")).thenReturn(Optional.empty());
+ jobWorker.initialize(contextManager);
+ jobWorker.initialize(contextManager);
+ assertFalse(workerInitialized.get());
+ serviceLoader.verify(() ->
TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"UNSUPPORTED"), times(2));
+ }
+
+ @Test
+ void assertInitializeFailureReleasesResourcesAndCanRetry() {
+ ClusterPersistRepositoryConfiguration repositoryConfig =
mockRepositoryConfiguration("FIXTURE");
+ CoordinatorRegistryCenter firstRegistryCenter =
mock(CoordinatorRegistryCenter.class);
+ CoordinatorRegistryCenter secondRegistryCenter =
mock(CoordinatorRegistryCenter.class);
+ serviceLoader.when(() ->
TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"FIXTURE")).thenReturn(Optional.of(provider));
+ when(provider.create(repositoryConfig,
STATISTICS_JOB_PATH)).thenReturn(firstRegistryCenter, secondRegistryCenter);
when(contextManager.getMetaDataContexts().getMetaData().getTemporaryProps()).thenReturn(new
TemporaryConfigurationProperties(new Properties()));
+ RuntimeException expected = new RuntimeException("expected");
+ AtomicBoolean failOnSchedule = new AtomicBoolean(true);
try (
- MockedConstruction<ZookeeperRegistryCenter>
registryCenterConstruction = mockConstruction(ZookeeperRegistryCenter.class);
- MockedConstruction<ScheduleJobBootstrap>
scheduleJobBootstrapConstruction = mockConstruction(ScheduleJobBootstrap.class);
+ MockedConstruction<ScheduleJobBootstrap>
scheduleJobBootstrapConstruction = mockConstruction(ScheduleJobBootstrap.class,
(mock, context) -> {
+ if (failOnSchedule.compareAndSet(true, false)) {
+ doThrow(expected).when(mock).schedule();
+ }
+ });
MockedConstruction<JobOperateAPIImpl>
jobOperateAPIConstruction = mockConstruction(JobOperateAPIImpl.class)) {
+ assertThat(assertThrows(RuntimeException.class, () ->
jobWorker.initialize(contextManager)), sameInstance(expected));
+ assertFalse(workerInitialized.get());
+
verify(scheduleJobBootstrapConstruction.constructed().get(0)).shutdown();
+ verify(firstRegistryCenter).close();
jobWorker.initialize(contextManager);
- verify(registryCenterConstruction.constructed().get(0)).init();
-
verify(scheduleJobBootstrapConstruction.constructed().get(0)).schedule();
+ assertTrue(workerInitialized.get());
+
verify(scheduleJobBootstrapConstruction.constructed().get(1)).schedule();
verify(jobOperateAPIConstruction.constructed().get(0)).trigger("statistics-collect");
+ jobWorker.destroy();
+
verify(scheduleJobBootstrapConstruction.constructed().get(1)).shutdown();
+ verify(secondRegistryCenter).close();
}
}
@Test
- void assertInitializeWithZooKeeperRepositoryUsingDefaultValues() {
-
when(contextManager.getComputeNodeInstanceContext().getModeConfiguration()).thenReturn(
- new ModeConfiguration("Cluster", new
ClusterPersistRepositoryConfiguration("ZooKeeper", "namespace",
"127.0.0.1:2181", new Properties())));
+ void assertPreserveInitializationFailureWhenCleanupFailed() {
+ ClusterPersistRepositoryConfiguration repositoryConfig =
mockRepositoryConfiguration("FIXTURE");
+ CoordinatorRegistryCenter registryCenter =
mock(CoordinatorRegistryCenter.class);
+ serviceLoader.when(() ->
TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"FIXTURE")).thenReturn(Optional.of(provider));
+ when(provider.create(repositoryConfig,
STATISTICS_JOB_PATH)).thenReturn(registryCenter);
when(contextManager.getMetaDataContexts().getMetaData().getTemporaryProps()).thenReturn(new
TemporaryConfigurationProperties(new Properties()));
- try (
- MockedConstruction<ZookeeperRegistryCenter>
registryCenterConstruction = mockConstruction(ZookeeperRegistryCenter.class);
- MockedConstruction<ScheduleJobBootstrap>
scheduleJobBootstrapConstruction = mockConstruction(ScheduleJobBootstrap.class);
- MockedConstruction<JobOperateAPIImpl>
jobOperateAPIConstruction = mockConstruction(JobOperateAPIImpl.class)) {
- jobWorker.initialize(contextManager);
- verify(registryCenterConstruction.constructed().get(0)).init();
-
verify(scheduleJobBootstrapConstruction.constructed().get(0)).schedule();
-
verify(jobOperateAPIConstruction.constructed().get(0)).trigger("statistics-collect");
+ RuntimeException expected = new RuntimeException("expected");
+ RuntimeException cleanupException = new RuntimeException("cleanup
failed");
+ try (MockedConstruction<ScheduleJobBootstrap>
scheduleJobBootstrapConstruction = mockConstruction(ScheduleJobBootstrap.class,
(mock, context) -> {
+ doThrow(expected).when(mock).schedule();
+ doThrow(cleanupException).when(mock).shutdown();
+ })) {
+ RuntimeException actual = assertThrows(RuntimeException.class, ()
-> jobWorker.initialize(contextManager));
+ assertThat(actual, sameInstance(expected));
+ assertThat(actual.getSuppressed(),
arrayContaining(cleanupException));
+ assertFalse(workerInitialized.get());
+
verify(scheduleJobBootstrapConstruction.constructed().get(0)).shutdown();
+ verify(registryCenter).close();
}
}
+ private ClusterPersistRepositoryConfiguration
mockRepositoryConfiguration(final String type) {
+ ClusterPersistRepositoryConfiguration result = new
ClusterPersistRepositoryConfiguration(type, "namespace", "127.0.0.1:2181", new
Properties());
+
when(contextManager.getComputeNodeInstanceContext().getModeConfiguration()).thenReturn(new
ModeConfiguration("Cluster", result));
+ return result;
+ }
+
@Test
void assertUpdateJobConfigurationWithNullContextManager() {
assertDoesNotThrow(() -> jobWorker.updateJobConfiguration());
diff --git a/mode/type/cluster/repository/provider/zookeeper/pom.xml
b/mode/type/cluster/repository/provider/zookeeper/pom.xml
index 338b8f6a4f7..17356601395 100644
--- a/mode/type/cluster/repository/provider/zookeeper/pom.xml
+++ b/mode/type/cluster/repository/provider/zookeeper/pom.xml
@@ -37,6 +37,16 @@
<artifactId>shardingsphere-cluster-mode-core</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.shardingsphere</groupId>
+ <artifactId>shardingsphere-schedule-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.shardingsphere.elasticjob</groupId>
+
<artifactId>elasticjob-registry-center-zookeeper-curator</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.curator</groupId>
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/registrycenter/elasticjob/CoordinatorRegistryCenterInitializer.java
b/mode/type/cluster/repository/provider/zookeeper/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/ZookeeperCoordinatorRegistryCenterProvider.java
similarity index 67%
rename from
kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/registrycenter/elasticjob/CoordinatorRegistryCenterInitializer.java
rename to
mode/type/cluster/repository/provider/zookeeper/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/ZookeeperCoordinatorRegistryCenterProvider.java
index 92b8c584fb2..d8be149740c 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/registrycenter/elasticjob/CoordinatorRegistryCenterInitializer.java
+++
b/mode/type/cluster/repository/provider/zookeeper/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/ZookeeperCoordinatorRegistryCenterProvider.java
@@ -15,42 +15,43 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.data.pipeline.core.registrycenter.elasticjob;
+package org.apache.shardingsphere.mode.repository.cluster.zookeeper;
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.infra.config.mode.ModeConfiguration;
import
org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
import
org.apache.shardingsphere.mode.repository.cluster.zookeeper.props.ZookeeperProperties;
import
org.apache.shardingsphere.mode.repository.cluster.zookeeper.props.ZookeeperPropertyKey;
-
-import java.util.Properties;
+import
org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider;
/**
- * {@linkplain CoordinatorRegistryCenter} initializer.
+ * ZooKeeper coordinator registry center provider.
*/
-public final class CoordinatorRegistryCenterInitializer {
+public final class ZookeeperCoordinatorRegistryCenterProvider implements
CoordinatorRegistryCenterProvider {
- /**
- * Create ZooKeeper registry center instance.
- *
- * @param modeConfig mode configuration
- * @param namespaceRelativePath namespace relative path
- * @return registry center instance
- */
- public CoordinatorRegistryCenter createZookeeperRegistryCenter(final
ModeConfiguration modeConfig, final String namespaceRelativePath) {
- ClusterPersistRepositoryConfiguration repositoryConfig =
(ClusterPersistRepositoryConfiguration) modeConfig.getRepository();
- // TODO Add registry center cache. Refer to
RegistryCenterFactory.createCoordinatorRegistryCenter
- CoordinatorRegistryCenter result = new
ZookeeperRegistryCenter(getZookeeperConfig(repositoryConfig,
namespaceRelativePath));
- result.init();
- return result;
+ @Override
+ public CoordinatorRegistryCenter create(final
ClusterPersistRepositoryConfiguration repositoryConfig, final String
namespaceRelativePath) {
+ CoordinatorRegistryCenter result = new
ZookeeperRegistryCenter(createConfiguration(repositoryConfig,
namespaceRelativePath));
+ try {
+ result.init();
+ return result;
+ // CHECKSTYLE:OFF
+ } catch (final RuntimeException ex) {
+ // CHECKSTYLE:ON
+ try {
+ result.close();
+ // CHECKSTYLE:OFF
+ } catch (final RuntimeException closeEx) {
+ // CHECKSTYLE:ON
+ ex.addSuppressed(closeEx);
+ }
+ throw ex;
+ }
}
- private ZookeeperConfiguration getZookeeperConfig(final
ClusterPersistRepositoryConfiguration repositoryConfig, final String
namespaceRelativePath) {
- // TODO Merge registry center code in ElasticJob and ShardingSphere
mode; Use SPI to load impl
- Properties props = repositoryConfig.getProps();
- ZookeeperProperties zookeeperProps = new ZookeeperProperties(props);
+ private ZookeeperConfiguration createConfiguration(final
ClusterPersistRepositoryConfiguration repositoryConfig, final String
namespaceRelativePath) {
+ ZookeeperProperties zookeeperProps = new
ZookeeperProperties(repositoryConfig.getProps());
String namespace = repositoryConfig.getNamespace() + (null ==
namespaceRelativePath ? "" : namespaceRelativePath);
ZookeeperConfiguration result = new
ZookeeperConfiguration(repositoryConfig.getServerLists(), namespace);
int retryIntervalMilliseconds =
zookeeperProps.getValue(ZookeeperPropertyKey.RETRY_INTERVAL_MILLISECONDS);
@@ -69,4 +70,9 @@ public final class CoordinatorRegistryCenterInitializer {
result.setDigest(zookeeperProps.getValue(ZookeeperPropertyKey.DIGEST));
return result;
}
+
+ @Override
+ public String getType() {
+ return "ZooKeeper";
+ }
}
diff --git
a/mode/type/cluster/repository/provider/zookeeper/src/main/resources/META-INF/services/org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider
b/mode/type/cluster/repository/provider/zookeeper/src/main/resources/META-INF/services/org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider
new file mode 100644
index 00000000000..6613a3d5b30
--- /dev/null
+++
b/mode/type/cluster/repository/provider/zookeeper/src/main/resources/META-INF/services/org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider
@@ -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.mode.repository.cluster.zookeeper.ZookeeperCoordinatorRegistryCenterProvider
diff --git
a/mode/type/cluster/repository/provider/zookeeper/src/test/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/ZookeeperCoordinatorRegistryCenterProviderTest.java
b/mode/type/cluster/repository/provider/zookeeper/src/test/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/ZookeeperCoordinatorRegistryCenterProviderTest.java
new file mode 100644
index 00000000000..5a00964eda2
--- /dev/null
+++
b/mode/type/cluster/repository/provider/zookeeper/src/test/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/ZookeeperCoordinatorRegistryCenterProviderTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.mode.repository.cluster.zookeeper;
+
+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.infra.spi.type.typed.TypedSPILoader;
+import
org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
+import
org.apache.shardingsphere.schedule.spi.CoordinatorRegistryCenterProvider;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.verify;
+
+class ZookeeperCoordinatorRegistryCenterProviderTest {
+
+ @Test
+ void assertIsSupportedWithoutCreatingRegistryCenter() {
+ try (MockedConstruction<ZookeeperRegistryCenter> construction =
mockConstruction(ZookeeperRegistryCenter.class)) {
+
assertTrue(TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"ZooKeeper").isPresent());
+
assertTrue(TypedSPILoader.findService(CoordinatorRegistryCenterProvider.class,
"zookeeper").isPresent());
+ assertTrue(construction.constructed().isEmpty());
+ }
+ }
+
+ @Test
+ void assertCreate() {
+ Properties props = new Properties();
+ props.setProperty("retryIntervalMilliseconds", "200");
+ props.setProperty("maxRetries", "4");
+ props.setProperty("timeToLiveSeconds", "5");
+ props.setProperty("operationTimeoutMilliseconds", "800");
+ props.setProperty("digest", "digest");
+ ClusterPersistRepositoryConfiguration repositoryConfig = new
ClusterPersistRepositoryConfiguration("ZooKeeper", "namespace",
"127.0.0.1:2181", props);
+ AtomicReference<ZookeeperConfiguration> configuration = new
AtomicReference<>();
+ try (
+ MockedConstruction<ZookeeperRegistryCenter> construction =
mockConstruction(ZookeeperRegistryCenter.class,
+ (mock, context) ->
configuration.set((ZookeeperConfiguration) context.arguments().get(0)))) {
+ CoordinatorRegistryCenter actual =
TypedSPILoader.getService(CoordinatorRegistryCenterProvider.class,
repositoryConfig.getType()).create(repositoryConfig, "/jobs");
+ assertThat(actual,
sameInstance(construction.constructed().get(0)));
+ verify(actual).init();
+ assertThat(configuration.get().getServerLists(),
is("127.0.0.1:2181"));
+ assertThat(configuration.get().getNamespace(),
is("namespace/jobs"));
+ assertThat(configuration.get().getBaseSleepTimeMilliseconds(),
is(200));
+ assertThat(configuration.get().getMaxRetries(), is(4));
+ assertThat(configuration.get().getMaxSleepTimeMilliseconds(),
is(800));
+ assertThat(configuration.get().getSessionTimeoutMilliseconds(),
is(5000));
+ assertThat(configuration.get().getConnectionTimeoutMilliseconds(),
is(800));
+ assertThat(configuration.get().getDigest(), is("digest"));
+ actual.close();
+ verify(actual).close();
+ }
+ }
+
+ @Test
+ void assertCreateWithZeroTimeoutsAndNullNamespaceRelativePath() {
+ Properties props = new Properties();
+ props.setProperty("timeToLiveSeconds", "0");
+ props.setProperty("operationTimeoutMilliseconds", "0");
+ ClusterPersistRepositoryConfiguration repositoryConfig = new
ClusterPersistRepositoryConfiguration("ZooKeeper", "namespace",
"127.0.0.1:2181", props);
+ AtomicReference<ZookeeperConfiguration> configuration = new
AtomicReference<>();
+ try (
+ MockedConstruction<ZookeeperRegistryCenter> ignored =
mockConstruction(ZookeeperRegistryCenter.class,
+ (mock, context) ->
configuration.set((ZookeeperConfiguration) context.arguments().get(0)))) {
+ TypedSPILoader.getService(CoordinatorRegistryCenterProvider.class,
repositoryConfig.getType()).create(repositoryConfig, null);
+ assertThat(configuration.get().getNamespace(), is("namespace"));
+ assertThat(configuration.get().getBaseSleepTimeMilliseconds(),
is(500));
+ assertThat(configuration.get().getMaxRetries(), is(3));
+ assertThat(configuration.get().getMaxSleepTimeMilliseconds(),
is(1500));
+ assertThat(configuration.get().getSessionTimeoutMilliseconds(),
is(0));
+ assertThat(configuration.get().getConnectionTimeoutMilliseconds(),
is(0));
+ assertThat(configuration.get().getDigest(), is(""));
+ }
+ }
+
+ @Test
+ void assertCloseRegistryCenterWhenInitializationFailed() {
+ ClusterPersistRepositoryConfiguration repositoryConfig = new
ClusterPersistRepositoryConfiguration("ZooKeeper", "namespace",
"127.0.0.1:2181", new Properties());
+ RuntimeException expected = new RuntimeException("expected");
+ try (
+ MockedConstruction<ZookeeperRegistryCenter> construction =
mockConstruction(ZookeeperRegistryCenter.class,
+ (mock, context) ->
doThrow(expected).when(mock).init())) {
+ assertThat(assertThrows(RuntimeException.class,
+ () ->
TypedSPILoader.getService(CoordinatorRegistryCenterProvider.class,
repositoryConfig.getType()).create(repositoryConfig, null)),
sameInstance(expected));
+ verify(construction.constructed().get(0)).close();
+ }
+ }
+
+ @Test
+ void assertPreserveInitializationFailureWhenCloseFailed() {
+ ClusterPersistRepositoryConfiguration repositoryConfig = new
ClusterPersistRepositoryConfiguration("ZooKeeper", "namespace",
"127.0.0.1:2181", new Properties());
+ RuntimeException expected = new RuntimeException("expected");
+ RuntimeException closeException = new RuntimeException("close failed");
+ try (MockedConstruction<ZookeeperRegistryCenter> ignored =
mockConstruction(ZookeeperRegistryCenter.class, (mock, context) -> {
+ doThrow(expected).when(mock).init();
+ doThrow(closeException).when(mock).close();
+ })) {
+ RuntimeException actual = assertThrows(RuntimeException.class,
+ () ->
TypedSPILoader.getService(CoordinatorRegistryCenterProvider.class,
repositoryConfig.getType()).create(repositoryConfig, null));
+ assertThat(actual, sameInstance(expected));
+ assertThat(actual.getSuppressed(),
arrayContaining(closeException));
+ }
+ }
+}
diff --git a/pom.xml b/pom.xml
index 54b3f192dae..ab0ca82d42f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -496,6 +496,11 @@
<artifactId>elasticjob-lite-lifecycle</artifactId>
<version>${elasticjob.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.shardingsphere.elasticjob</groupId>
+
<artifactId>elasticjob-registry-center-zookeeper-curator</artifactId>
+ <version>${elasticjob.version}</version>
+ </dependency>
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
