Repository: apache/shardingsphere-elasticjob-lite (java)

Issue:
Disable job in startup is not in effect sometimes
The reason is curator save disable status async, and sometimes not persist to ZooKeeper in time before return. 
It will cause Elastic Job judge the server as enabled server.

We should load server status after make sure the server status persist in to ZooKeeper.

- [x] Add ENABLE status
- [x] Judge if the server status is empty sleep and reload again

Pull request description:
Fix issue for disable job in startup is not in effect sometimes
Fixes #946.


Patch:
diff --git a/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerService.java b/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerService.java
index 7985a7e22..fb73b2cf7 100644
--- a/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerService.java
+++ b/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerService.java
@@ -17,10 +17,12 @@
 
 package org.apache.shardingsphere.elasticjob.lite.internal.server;
 
+import com.google.common.base.Strings;
 import org.apache.shardingsphere.elasticjob.lite.internal.instance.InstanceNode;
 import org.apache.shardingsphere.elasticjob.lite.internal.schedule.JobRegistry;
 import org.apache.shardingsphere.elasticjob.lite.internal.storage.JobNodeStorage;
 import org.apache.shardingsphere.elasticjob.lite.reg.base.CoordinatorRegistryCenter;
+import org.apache.shardingsphere.elasticjob.lite.util.concurrent.BlockUtils;
 
 import java.util.List;
 
@@ -48,7 +50,7 @@ public final class ServerService {
      */
     public void persistOnline(final boolean enabled) {
         if (!JobRegistry.getInstance().isShutdown(jobName)) {
-            jobNodeStorage.fillJobNode(serverNode.getServerNode(JobRegistry.getInstance().getJobInstance(jobName).getIp()), enabled ? "" : ServerStatus.DISABLED.name());
+            jobNodeStorage.fillJobNode(serverNode.getServerNode(JobRegistry.getInstance().getJobInstance(jobName).getIp()), enabled ? ServerStatus.ENABLED.name() : ServerStatus.DISABLED.name());
         }
     }
     
@@ -93,6 +95,11 @@ public final class ServerService {
      * @return is server enabled or not
      */
     public boolean isEnableServer(final String ip) {
-        return !ServerStatus.DISABLED.name().equals(jobNodeStorage.getJobNodeData(serverNode.getServerNode(ip)));
+        String serverStatus = jobNodeStorage.getJobNodeData(serverNode.getServerNode(ip));
+        while (Strings.isNullOrEmpty(serverStatus)) {
+            BlockUtils.waitingShortTime();
+            serverStatus = jobNodeStorage.getJobNodeData(serverNode.getServerNode(ip));
+        }
+        return !ServerStatus.DISABLED.name().equals(serverStatus);
     }
 }
diff --git a/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerStatus.java b/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerStatus.java
index 7144f2f37..781da1988 100644
--- a/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerStatus.java
+++ b/elastic-job-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerStatus.java
@@ -22,5 +22,5 @@ package org.apache.shardingsphere.elasticjob.lite.internal.server;
  */
 public enum ServerStatus {
     
-    DISABLED
+    ENABLED, DISABLED
 }


Test patch:
diff --git a/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/integrate/BaseIntegrateTest.java b/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/integrate/BaseIntegrateTest.java
index 8b716733c..7a4d3fe02 100644
--- a/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/integrate/BaseIntegrateTest.java
+++ b/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/integrate/BaseIntegrateTest.java
@@ -151,7 +151,7 @@ public abstract class BaseIntegrateTest {
             }
             regCenter.persist("/" + jobName + "/servers/" + JobRegistry.getInstance().getJobInstance(jobName).getIp(), "");
         } else {
-            assertThat(regCenter.get("/" + jobName + "/servers/" + JobRegistry.getInstance().getJobInstance(jobName).getIp()), is(""));
+            assertThat(regCenter.get("/" + jobName + "/servers/" + JobRegistry.getInstance().getJobInstance(jobName).getIp()), is(ServerStatus.ENABLED.name()));
             assertThat(regCenter.get("/" + jobName + "/leader/election/instance"), is(JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId()));
         }
         assertTrue(regCenter.isExisted("/" + jobName + "/instances/" + JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId()));
diff --git a/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/integrate/DisabledJobIntegrateTest.java b/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/integrate/DisabledJobIntegrateTest.java
index b5d3fb7c8..88b9aeb84 100644
--- a/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/integrate/DisabledJobIntegrateTest.java
+++ b/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/integrate/DisabledJobIntegrateTest.java
@@ -37,9 +37,7 @@ public abstract class DisabledJobIntegrateTest extends BaseIntegrateTest {
     
     @Test
     public final void assertJobRunning() {
+        BlockUtils.waitingShortTime();
         assertRegCenterCommonInfoWithDisabled();
-        while (!FooSimpleJob.isCompleted()) {
-            BlockUtils.waitingShortTime();
-        }
     }
 }
diff --git a/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerServiceTest.java b/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerServiceTest.java
index 095a317f7..f95f60d0b 100644
--- a/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerServiceTest.java
+++ b/elastic-job-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/server/ServerServiceTest.java
@@ -82,7 +82,7 @@ public final class ServerServiceTest {
         JobRegistry.getInstance().registerRegistryCenter("test_job", regCenter);
         JobRegistry.getInstance().registerJob("test_job", jobScheduleController);
         serverService.persistOnline(true);
-        verify(jobNodeStorage).fillJobNode("servers/127.0.0.1", "");
+        verify(jobNodeStorage).fillJobNode("servers/127.0.0.1", ServerStatus.ENABLED.name());
         JobRegistry.getInstance().shutdown("test_job");
     }
     
@@ -90,6 +90,8 @@ public final class ServerServiceTest {
     public void assertHasAvailableServers() {
         when(jobNodeStorage.getJobNodeChildrenKeys("servers")).thenReturn(Arrays.asList("127.0.0.1", "127.0.0.2", "127.0.0.3"));
         when(jobNodeStorage.getJobNodeData("servers/127.0.0.1")).thenReturn(ServerStatus.DISABLED.name());
+        when(jobNodeStorage.getJobNodeData("servers/127.0.0.2")).thenReturn(ServerStatus.ENABLED.name());
+        when(jobNodeStorage.getJobNodeData("servers/127.0.0.3")).thenReturn(ServerStatus.ENABLED.name());
         when(jobNodeStorage.getJobNodeChildrenKeys("instances")).thenReturn(Collections.singletonList("127.0.0.3@-@0"));
         assertTrue(serverService.hasAvailableServers());
     }
@@ -97,8 +99,8 @@ public final class ServerServiceTest {
     @Test
     public void assertHasNotAvailableServers() {
         when(jobNodeStorage.getJobNodeChildrenKeys("servers")).thenReturn(Arrays.asList("127.0.0.1", "127.0.0.2"));
+        when(jobNodeStorage.getJobNodeData("servers/127.0.0.1")).thenReturn(ServerStatus.DISABLED.name());
         when(jobNodeStorage.getJobNodeData("servers/127.0.0.2")).thenReturn(ServerStatus.DISABLED.name());
-        when(jobNodeStorage.getJobNodeChildrenKeys("instances")).thenReturn(Arrays.asList("127.0.0.2@-@0", "127.0.0.2@-@1"));
         assertFalse(serverService.hasAvailableServers());
     }
     
@@ -111,23 +113,26 @@ public final class ServerServiceTest {
     @Test
     public void assertIsNotAvailableServerWithoutOnlineInstances() {
         when(jobNodeStorage.getJobNodeChildrenKeys("instances")).thenReturn(Collections.singletonList("127.0.0.2@-@0"));
+        when(jobNodeStorage.getJobNodeData("servers/127.0.0.1")).thenReturn(ServerStatus.ENABLED.name());
         assertFalse(serverService.isAvailableServer("127.0.0.1"));
     }
     
     @Test
     public void assertIsAvailableServer() {
         when(jobNodeStorage.getJobNodeChildrenKeys("instances")).thenReturn(Collections.singletonList("127.0.0.1@-@0"));
+        when(jobNodeStorage.getJobNodeData("servers/127.0.0.1")).thenReturn(ServerStatus.ENABLED.name());
         assertTrue(serverService.isAvailableServer("127.0.0.1"));
     }
     
     @Test
     public void assertIsNotEnableServer() {
-        when(jobNodeStorage.getJobNodeData("servers/127.0.0.1")).thenReturn(ServerStatus.DISABLED.name());
+        when(jobNodeStorage.getJobNodeData("servers/127.0.0.1")).thenReturn("", ServerStatus.DISABLED.name());
         assertFalse(serverService.isEnableServer("127.0.0.1"));
     }
     
     @Test
     public void assertIsEnableServer() {
+        when(jobNodeStorage.getJobNodeData("servers/127.0.0.1")).thenReturn("", ServerStatus.ENABLED.name());
         assertTrue(serverService.isEnableServer("127.0.0.1"));
     }
 }