Repository: apache/shardingsphere-elasticjob
Issue: Resuming Transient Job is abnormal

### Which version of Elastic-Job do you using?
2.1.4

### Expected behavior
Disable the job and then resume the job, then triggered by cron expression. 

### Actual behavior
Disable the job and then resume the job, immediately triggered several times.

### Steps to reproduce the behavior
Disable the transient job(cron 0/30 * * * * ?) with "misfire" parameter, wait several minutes, then resume it.


Changed files:
elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApi.java
elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeServiceTest.java
elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListener.java
elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeService.java
elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudAppRestfulApi.java
elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListenerTest.java
elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApiTest.java
elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/TransientProducerScheduler.java
elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/ProducerManager.java

diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListener.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListener.java
index 0f9e58ae7..bccbb183a 100644
--- a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListener.java
+++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListener.java
@@ -70,7 +70,7 @@ public final class CloudJobConfigurationListener implements TreeCacheListener {
             if (!jobConfig.getTypeConfig().getCoreConfig().isMisfire()) {
                 readyService.setMisfireDisabled(jobConfig.getJobName());
             }
-            producerManager.reschedule(jobConfig);
+            producerManager.reschedule(jobConfig.getJobName());
         } else if (isJobConfigNode(event, path, Type.NODE_REMOVED)) {
             String jobName = path.substring(CloudJobConfigurationNode.ROOT.length() + 1, path.length());
             producerManager.unschedule(jobName);
diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeService.java
index 527c58e1a..d27c36c53 100644
--- a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeService.java
+++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeService.java
@@ -172,12 +172,19 @@ public final class FacadeService {
         if (!jobConfigOptional.isPresent()) {
             return;
         }
+        if (isDisable(jobConfigOptional.get())) {
+            return;
+        }
         CloudJobConfiguration jobConfig = jobConfigOptional.get();
         if (jobConfig.getTypeConfig().getCoreConfig().isFailover() || CloudJobExecutionType.DAEMON == jobConfig.getJobExecutionType()) {
             failoverService.add(taskContext);
         }
     }
     
+    private boolean isDisable(final CloudJobConfiguration jobConfiguration) {
+        return disableAppService.isDisabled(jobConfiguration.getAppName()) || disableJobService.isDisabled(jobConfiguration.getJobName());
+    }
+    
     /**
      * 将瞬时作业放入待执行队列.
      *
@@ -223,6 +230,13 @@ public final class FacadeService {
      * @param jobName 作业名称
      */
     public void addDaemonJobToReadyQueue(final String jobName) {
+        Optional<CloudJobConfiguration> jobConfigOptional = jobConfigService.load(jobName);
+        if (!jobConfigOptional.isPresent()) {
+            return;
+        }
+        if (isDisable(jobConfigOptional.get())) {
+            return;
+        }
         readyService.addDaemon(jobName);
     }
     
diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/ProducerManager.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/ProducerManager.java
index 71d4c6d37..e681e656e 100644
--- a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/ProducerManager.java
+++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/ProducerManager.java
@@ -117,7 +117,7 @@ public final class ProducerManager {
             throw new JobConfigurationException("Cannot found job '%s', please register first.", jobConfig.getJobName());
         }
         configService.update(jobConfig);
-        reschedule(jobConfig);
+        reschedule(jobConfig.getJobName());
     }
     
     /**
@@ -130,7 +130,6 @@ public final class ProducerManager {
         if (jobConfig.isPresent()) {
             disableJobService.remove(jobName);
             configService.remove(jobName);
-            transientProducerScheduler.deregister(jobConfig.get());
         }
         unschedule(jobName);
     }
@@ -162,16 +161,23 @@ public final class ProducerManager {
         }
         runningService.remove(jobName);
         readyService.remove(Lists.newArrayList(jobName));
+        Optional<CloudJobConfiguration> jobConfig = configService.load(jobName);
+        if (jobConfig.isPresent()) {
+            transientProducerScheduler.deregister(jobConfig.get());
+        }
     }
     
     /**
      * 重新调度作业.
      *
-     * @param jobConfig 作业配置
+     * @param jobName 作业名称
      */
-    public void reschedule(final CloudJobConfiguration jobConfig) {
-        unschedule(jobConfig.getJobName());
-        schedule(jobConfig);
+    public void reschedule(final String jobName) {
+        unschedule(jobName);
+        Optional<CloudJobConfiguration> jobConfig = configService.load(jobName);
+        if (jobConfig.isPresent()) {
+            schedule(jobConfig.get());
+        }
     }
     
     /**
diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/TransientProducerScheduler.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/TransientProducerScheduler.java
index bfb793cc0..93d379f9d 100644
--- a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/TransientProducerScheduler.java
+++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/producer/TransientProducerScheduler.java
@@ -112,7 +112,7 @@ final class TransientProducerScheduler {
         return TriggerBuilder.newTrigger().withIdentity(cron).withSchedule(CronScheduleBuilder.cronSchedule(cron).withMisfireHandlingInstructionDoNothing()).build();
     }
     
-    void deregister(final CloudJobConfiguration jobConfig) {
+    synchronized void deregister(final CloudJobConfiguration jobConfig) {
         repository.remove(jobConfig.getJobName());
         String cron = jobConfig.getTypeConfig().getCoreConfig().getCron();
         if (!repository.containsKey(buildJobKey(cron))) {
diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudAppRestfulApi.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudAppRestfulApi.java
index 5c2c7f6a2..9d849510e 100644
--- a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudAppRestfulApi.java
+++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudAppRestfulApi.java
@@ -186,6 +186,11 @@ public final class CloudAppRestfulApi {
     public void enable(@PathParam("appName") final String appName) throws JSONException {
         if (appConfigService.load(appName).isPresent()) {
             disableAppService.remove(appName);
+            for (CloudJobConfiguration each : jobConfigService.loadAll()) {
+                if (appName.equals(each.getAppName())) {
+                    producerManager.reschedule(each.getJobName());
+                }
+            }
         }
     }
     
diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApi.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApi.java
index 2ad20f6db..2505b0435 100644
--- a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApi.java
+++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApi.java
@@ -184,8 +184,10 @@ public final class CloudJobRestfulApi {
     @DELETE
     @Path("/{jobName}/disable")
     public void enable(@PathParam("jobName") final String jobName) throws JSONException {
-        if (configService.load(jobName).isPresent()) {
+        Optional<CloudJobConfiguration> configOptional = configService.load(jobName);
+        if (configOptional.isPresent()) {
             facadeService.enableJob(jobName);
+            producerManager.reschedule(jobName);
         }
     }
     
diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListenerTest.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListenerTest.java
index 99f7ae93e..e8dc9213b 100644
--- a/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListenerTest.java
+++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationListenerTest.java
@@ -58,7 +58,7 @@ public final class CloudJobConfigurationListenerTest {
     public void assertChildEventWhenDataIsNull() throws Exception {
         cloudJobConfigurationListener.childEvent(null, new TreeCacheEvent(TreeCacheEvent.Type.NODE_ADDED, null));
         verify(producerManager, times(0)).schedule(ArgumentMatchers.<CloudJobConfiguration>any());
-        verify(producerManager, times(0)).reschedule(ArgumentMatchers.<CloudJobConfiguration>any());
+        verify(producerManager, times(0)).reschedule(ArgumentMatchers.<String>any());
         verify(producerManager, times(0)).unschedule(ArgumentMatchers.<String>any());
     }
     
@@ -66,7 +66,7 @@ public final class CloudJobConfigurationListenerTest {
     public void assertChildEventWhenIsNotConfigPath() throws Exception {
         cloudJobConfigurationListener.childEvent(null, new TreeCacheEvent(TreeCacheEvent.Type.NODE_UPDATED, new ChildData("/other/test_job", null, "".getBytes())));
         verify(producerManager, times(0)).schedule(ArgumentMatchers.<CloudJobConfiguration>any());
-        verify(producerManager, times(0)).reschedule(ArgumentMatchers.<CloudJobConfiguration>any());
+        verify(producerManager, times(0)).reschedule(ArgumentMatchers.<String>any());
         verify(producerManager, times(0)).unschedule(ArgumentMatchers.<String>any());
     }
     
@@ -74,7 +74,7 @@ public final class CloudJobConfigurationListenerTest {
     public void assertChildEventWhenIsRootConfigPath() throws Exception {
         cloudJobConfigurationListener.childEvent(null, new TreeCacheEvent(TreeCacheEvent.Type.NODE_REMOVED, new ChildData("/config/job", null, "".getBytes())));
         verify(producerManager, times(0)).schedule(ArgumentMatchers.<CloudJobConfiguration>any());
-        verify(producerManager, times(0)).reschedule(ArgumentMatchers.<CloudJobConfiguration>any());
+        verify(producerManager, times(0)).reschedule(ArgumentMatchers.<String>any());
         verify(producerManager, times(0)).unschedule(ArgumentMatchers.<String>any());
     }
     
@@ -82,7 +82,7 @@ public final class CloudJobConfigurationListenerTest {
     public void assertChildEventWhenStateIsAddAndIsConfigPathAndInvalidData() throws Exception {
         cloudJobConfigurationListener.childEvent(null, new TreeCacheEvent(TreeCacheEvent.Type.NODE_ADDED, new ChildData("/config/job/test_job", null, "".getBytes())));
         verify(producerManager, times(0)).schedule(ArgumentMatchers.<CloudJobConfiguration>any());
-        verify(producerManager, times(0)).reschedule(ArgumentMatchers.<CloudJobConfiguration>any());
+        verify(producerManager, times(0)).reschedule(ArgumentMatchers.<String>any());
         verify(producerManager, times(0)).unschedule(ArgumentMatchers.<String>any());
     }
     
@@ -96,7 +96,7 @@ public final class CloudJobConfigurationListenerTest {
     public void assertChildEventWhenStateIsUpdateAndIsConfigPathAndTransientJob() throws Exception {
         cloudJobConfigurationListener.childEvent(null, new TreeCacheEvent(TreeCacheEvent.Type.NODE_UPDATED, new ChildData("/config/job/test_job", null, CloudJsonConstants.getJobJson().getBytes())));
         verify(readyService, times(0)).remove(Collections.singletonList("test_job"));
-        verify(producerManager).reschedule(ArgumentMatchers.<CloudJobConfiguration>any());
+        verify(producerManager).reschedule(ArgumentMatchers.<String>any());
     }
     
     @Test
@@ -104,7 +104,7 @@ public final class CloudJobConfigurationListenerTest {
         cloudJobConfigurationListener.childEvent(null, new TreeCacheEvent(TreeCacheEvent.Type.NODE_UPDATED, 
                 new ChildData("/config/job/test_job", null, CloudJsonConstants.getJobJson(CloudJobExecutionType.DAEMON).getBytes())));
         verify(readyService).remove(Collections.singletonList("test_job"));
-        verify(producerManager).reschedule(ArgumentMatchers.<CloudJobConfiguration>any());
+        verify(producerManager).reschedule(ArgumentMatchers.<String>any());
     }
     
     @Test
@@ -112,7 +112,7 @@ public final class CloudJobConfigurationListenerTest {
         cloudJobConfigurationListener.childEvent(null, new TreeCacheEvent(TreeCacheEvent.Type.NODE_UPDATED,
                 new ChildData("/config/job/test_job", null, CloudJsonConstants.getJobJson(false).getBytes())));
         verify(readyService).setMisfireDisabled("test_job");
-        verify(producerManager).reschedule(ArgumentMatchers.<CloudJobConfiguration>any());
+        verify(producerManager).reschedule(ArgumentMatchers.<String>any());
     }
     
     @Test
diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeServiceTest.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeServiceTest.java
index 35e44ccde..50cc09c4f 100644
--- a/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeServiceTest.java
+++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeServiceTest.java
@@ -222,6 +222,7 @@ public final class FacadeServiceTest {
     
     @Test
     public void assertAddDaemonJobToReadyQueue() {
+        when(jobConfigService.load("test_job")).thenReturn(Optional.of(CloudJobConfigurationBuilder.createCloudJobConfiguration("test_job")));
         facadeService.addDaemonJobToReadyQueue("test_job");
         verify(readyService).addDaemon("test_job");
     }
diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApiTest.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApiTest.java
index e1c902886..1956a5da2 100644
--- a/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApiTest.java
+++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/test/java/com/dangdang/ddframe/job/cloud/scheduler/restful/CloudJobRestfulApiTest.java
@@ -106,7 +106,7 @@ public final class CloudJobRestfulApiTest extends AbstractCloudRestfulApiTest {
     public void assertDeregister() throws Exception {
         when(getRegCenter().isExisted("/config/job/test_job")).thenReturn(false);
         assertThat(sentRequest("http://127.0.0.1:19000/api/job/deregister", "DELETE", "test_job"), is(204));
-        verify(getRegCenter(), times(2)).get("/config/job/test_job");
+        verify(getRegCenter(), times(3)).get("/config/job/test_job");
     }
     
     @Test