You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2020/07/01 13:40:35 UTC
[shardingsphere-elasticjob-lite] branch master updated: Fixed
FailoverListenerManager cause NPE(#860) (#881)
This is an automated email from the ASF dual-hosted git repository.
zhangliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere-elasticjob-lite.git
The following commit(s) were added to refs/heads/master by this push:
new d080e33 Fixed FailoverListenerManager cause NPE(#860) (#881)
d080e33 is described below
commit d080e331a3b867724088b9125a9725e06a7286da
Author: zhangzc <34...@qq.com>
AuthorDate: Wed Jul 1 21:40:20 2020 +0800
Fixed FailoverListenerManager cause NPE(#860) (#881)
Co-authored-by: zhangzc <zh...@gmail.com>
---
.../lite/internal/config/ConfigurationService.java | 1 -
.../lite/internal/failover/FailoverListenerManager.java | 2 +-
.../lite/internal/failover/FailoverListenerManagerTest.java | 13 +++++++++++++
3 files changed, 14 insertions(+), 2 deletions(-)
diff --git a/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/config/ConfigurationService.java b/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/config/ConfigurationService.java
index a616819..fb38c25 100644
--- a/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/config/ConfigurationService.java
+++ b/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/config/ConfigurationService.java
@@ -56,7 +56,6 @@ public final class ConfigurationService {
} else {
result = jobNodeStorage.getJobNodeDataDirectly(ConfigurationNode.ROOT);
}
- // TODO investigate why sometimes result is null
return YamlEngine.unmarshal(result, YamlJobConfiguration.class).toJobConfiguration();
}
diff --git a/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/failover/FailoverListenerManager.java b/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/failover/FailoverListenerManager.java
index 20ed326..aaa24a6 100644
--- a/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/failover/FailoverListenerManager.java
+++ b/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/failover/FailoverListenerManager.java
@@ -74,7 +74,7 @@ public final class FailoverListenerManager extends AbstractListenerManager {
@Override
protected void dataChanged(final String path, final Type eventType, final String data) {
- if (isFailoverEnabled() && Type.NODE_REMOVED == eventType && instanceNode.isInstancePath(path)) {
+ if (!JobRegistry.getInstance().isShutdown(jobName) && isFailoverEnabled() && Type.NODE_REMOVED == eventType && instanceNode.isInstancePath(path)) {
String jobInstanceId = path.substring(instanceNode.getInstanceFullPath().length() + 1);
if (jobInstanceId.equals(JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId())) {
return;
diff --git a/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/failover/FailoverListenerManagerTest.java b/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/failover/FailoverListenerManagerTest.java
index 1eb4d92..f292c46 100644
--- a/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/failover/FailoverListenerManagerTest.java
+++ b/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/failover/FailoverListenerManagerTest.java
@@ -25,6 +25,7 @@ import org.apache.shardingsphere.elasticjob.lite.handler.sharding.JobInstance;
import org.apache.shardingsphere.elasticjob.lite.internal.config.ConfigurationService;
import org.apache.shardingsphere.elasticjob.lite.internal.listener.AbstractJobListener;
import org.apache.shardingsphere.elasticjob.lite.internal.schedule.JobRegistry;
+import org.apache.shardingsphere.elasticjob.lite.internal.schedule.JobScheduleController;
import org.apache.shardingsphere.elasticjob.lite.internal.sharding.ShardingService;
import org.apache.shardingsphere.elasticjob.lite.internal.storage.JobNodeStorage;
import org.apache.shardingsphere.elasticjob.lite.util.ReflectionUtils;
@@ -47,6 +48,9 @@ public final class FailoverListenerManagerTest {
@Mock
private JobNodeStorage jobNodeStorage;
+
+ @Mock
+ private JobScheduleController jobScheduleController;
@Mock
private ConfigurationService configService;
@@ -81,21 +85,28 @@ public final class FailoverListenerManagerTest {
@Test
public void assertJobCrashedJobListenerWhenIsNotNodeRemoved() {
+ JobRegistry.getInstance().addJobInstance("test_job", new JobInstance("127.0.0.1@-@0"));
+ JobRegistry.getInstance().registerJob("test_job", jobScheduleController);
when(configService.load(true)).thenReturn(JobConfiguration.newBuilder("test_job", JobType.SIMPLE, 3).cron("0/1 * * * * ?").failover(true).build());
failoverListenerManager.new JobCrashedJobListener().dataChanged("/test_job/instances/127.0.0.1@-@0", Type.NODE_ADDED, "");
verify(failoverService, times(0)).failoverIfNecessary();
+ JobRegistry.getInstance().shutdown("test_job");
}
@Test
public void assertJobCrashedJobListenerWhenIsNotInstancesPath() {
+ JobRegistry.getInstance().addJobInstance("test_job", new JobInstance("127.0.0.1@-@0"));
+ JobRegistry.getInstance().registerJob("test_job", jobScheduleController);
when(configService.load(true)).thenReturn(JobConfiguration.newBuilder("test_job", JobType.SIMPLE, 3).cron("0/1 * * * * ?").failover(true).build());
failoverListenerManager.new JobCrashedJobListener().dataChanged("/test_job/other/127.0.0.1@-@0", Type.NODE_REMOVED, "");
verify(failoverService, times(0)).failoverIfNecessary();
+ JobRegistry.getInstance().shutdown("test_job");
}
@Test
public void assertJobCrashedJobListenerWhenIsSameInstance() {
JobRegistry.getInstance().addJobInstance("test_job", new JobInstance("127.0.0.1@-@0"));
+ JobRegistry.getInstance().registerJob("test_job", jobScheduleController);
when(configService.load(true)).thenReturn(JobConfiguration.newBuilder("test_job", JobType.SIMPLE, 3).cron("0/1 * * * * ?").failover(true).build());
failoverListenerManager.new JobCrashedJobListener().dataChanged("/test_job/instances/127.0.0.1@-@0", Type.NODE_REMOVED, "");
verify(failoverService, times(0)).failoverIfNecessary();
@@ -105,6 +116,7 @@ public final class FailoverListenerManagerTest {
@Test
public void assertJobCrashedJobListenerWhenIsOtherInstanceCrashed() {
JobRegistry.getInstance().addJobInstance("test_job", new JobInstance("127.0.0.1@-@0"));
+ JobRegistry.getInstance().registerJob("test_job", jobScheduleController);
when(configService.load(true)).thenReturn(JobConfiguration.newBuilder("test_job", JobType.SIMPLE, 3).cron("0/1 * * * * ?").failover(true).build());
when(shardingService.getShardingItems("127.0.0.1@-@1")).thenReturn(Arrays.asList(0, 2));
failoverListenerManager.new JobCrashedJobListener().dataChanged("/test_job/instances/127.0.0.1@-@1", Type.NODE_REMOVED, "");
@@ -117,6 +129,7 @@ public final class FailoverListenerManagerTest {
@Test
public void assertJobCrashedJobListenerWhenIsOtherFailoverInstanceCrashed() {
JobRegistry.getInstance().addJobInstance("test_job", new JobInstance("127.0.0.1@-@0"));
+ JobRegistry.getInstance().registerJob("test_job", jobScheduleController);
when(configService.load(true)).thenReturn(JobConfiguration.newBuilder("test_job", JobType.SIMPLE, 3).cron("0/1 * * * * ?").failover(true).build());
when(failoverService.getFailoverItems("127.0.0.1@-@1")).thenReturn(Collections.singletonList(1));
failoverListenerManager.new JobCrashedJobListener().dataChanged("/test_job/instances/127.0.0.1@-@1", Type.NODE_REMOVED, "");