#43

apache · Feb 1, 2016 · d8a94d8 · d8a94d8
1 parent 5128e30
commit d8a94d8
Show file tree

Hide file tree

Showing 9 changed files with 38 additions and 18 deletions.
diff --git a/...-core/src/main/java/com/dangdang/ddframe/job/internal/job/AbstractDataFlowElasticJob.java b/...-core/src/main/java/com/dangdang/ddframe/job/internal/job/AbstractDataFlowElasticJob.java
@@ -20,11 +20,14 @@
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 
-import lombok.extern.slf4j.Slf4j;
+import org.quartz.JobExecutionException;
 
 import com.dangdang.ddframe.job.api.DataFlowElasticJob;
+import com.dangdang.ddframe.job.exception.JobException;
 import com.dangdang.ddframe.job.internal.statistics.ProcessCountStatistics;
 
+import lombok.extern.slf4j.Slf4j;
+
 /**
  * 用于处理数据流程的作业抽象类.
  * 
@@ -42,7 +45,13 @@ public final void updateOffset(final int item, final String offset) {
         getOffsetService().updateOffset(item, offset);
     }
 
+    @Override
+    public void handleJobExecutionException(final JobExecutionException jobExecutionException) throws JobExecutionException {
+        log.error("Elastic job: exception occur in job processing...", jobExecutionException.getCause());
+    }
+
     protected final void processDataWithStatistics(final C shardingContext, final List<T> data) {
+        Exception firstException = null;
         for (T each : data) {
             boolean isSuccess = false;
             try {
@@ -51,7 +60,9 @@ protected final void processDataWithStatistics(final C shardingContext, final Li
             } catch (final Exception ex) {
             // CHECKSTYLE:ON
                 ProcessCountStatistics.incrementProcessFailureCount(shardingContext.getJobName());
-                log.error("Elastic job: exception occur in job processing...", ex);
+                if (null == firstException) {
+                    firstException = ex;
+                }
                 continue;
             }
             if (isSuccess) {
@@ -60,6 +71,9 @@ protected final void processDataWithStatistics(final C shardingContext, final Li
                 ProcessCountStatistics.incrementProcessFailureCount(shardingContext.getJobName());
             }
         }
+        if (null != firstException) {
+            throw new JobException(firstException);
+        }
     }
 
     protected final void latchAwait(final CountDownLatch latch) {

diff --git a/...core/src/main/java/com/dangdang/ddframe/job/plugin/job/type/AbstractSimpleElasticJob.java b/...core/src/main/java/com/dangdang/ddframe/job/plugin/job/type/AbstractSimpleElasticJob.java
@@ -19,6 +19,8 @@
 
 import lombok.extern.slf4j.Slf4j;
 
+import org.quartz.JobExecutionException;
+
 import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;
 import com.dangdang.ddframe.job.internal.job.AbstractElasticJob;
 
@@ -36,13 +38,12 @@ public abstract class AbstractSimpleElasticJob extends AbstractElasticJob {
 
     @Override
     protected final void executeJob(final JobExecutionMultipleShardingContext shardingContext) {
-        try {
-            process(shardingContext);
-         // CHECKSTYLE:OFF
-        } catch (final Exception ex) {
-         // CHECKSTYLE:ON
-            log.error("Elastic job: exception occur in job processing...", ex);
-        }
+        process(shardingContext);
+    }
+
+    @Override
+    public void handleJobExecutionException(final JobExecutionException jobExecutionException) throws JobExecutionException {
+        log.error("Elastic job: exception occur in job processing...", jobExecutionException.getCause());
     }
 
     /**

diff --git a/...ic-job-core/src/main/java/com/dangdang/ddframe/reg/zookeeper/ZookeeperRegistryCenter.java b/...ic-job-core/src/main/java/com/dangdang/ddframe/reg/zookeeper/ZookeeperRegistryCenter.java
@@ -139,6 +139,7 @@ private Properties loadLocalProperties() {
     public void close() {
         for (Entry<String, TreeCache> each : caches.entrySet()) {
             each.getValue().close();
+
         }
         waitForCacheClose();
         CloseableUtils.closeQuietly(client);

diff --git a/...tic-job-core/src/test/java/com/dangdang/ddframe/job/integrate/AbstractBaseStdJobTest.java b/...tic-job-core/src/test/java/com/dangdang/ddframe/job/integrate/AbstractBaseStdJobTest.java
@@ -23,7 +23,6 @@
 import static org.junit.Assert.assertTrue;
 
 import org.junit.After;
-import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.quartz.SchedulerException;
@@ -91,11 +90,6 @@ public static void init() {
         REG_CENTER.init();
     }
 
-    @AfterClass
-    public static void destory() {
-        REG_CENTER.close();
-    }
-
     @Before
     public void setUp() {
         ProcessCountStatistics.reset(jobName);

diff --git a/...e/dataflow/throughput/StreamingThroughputDataFlowElasticJobForExecuteThrowsException.java b/...e/dataflow/throughput/StreamingThroughputDataFlowElasticJobForExecuteThrowsException.java
@@ -20,6 +20,8 @@
 import java.util.Arrays;
 import java.util.List;
 
+import org.quartz.JobExecutionException;
+
 import lombok.Getter;
 
 import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;
@@ -50,6 +52,10 @@ public boolean isStreamingProcess() {
         return true;
     }
 
+    @Override
+    public void handleJobExecutionException(final JobExecutionException jobExecutionException) throws JobExecutionException {
+    }
+
     public static void reset() {
         completed = false;
     }

diff --git a/...e/std/dataflow/throughput/StreamingThroughputDataFlowElasticJobForExecuteFailureTest.java b/...e/std/dataflow/throughput/StreamingThroughputDataFlowElasticJobForExecuteFailureTest.java
@@ -18,6 +18,7 @@
 package com.dangdang.ddframe.job.integrate.std.dataflow.throughput;
 
 import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.not;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
@@ -49,6 +50,6 @@ public void assertJobInit() {
         }
         assertTrue(REG_CENTER.isExisted("/" + getJobName() + "/execution"));
         assertThat(ProcessCountStatistics.getProcessSuccessCount(getJobName()), is(0));
-        assertThat(ProcessCountStatistics.getProcessFailureCount(getJobName()), is(1));
+        assertThat(ProcessCountStatistics.getProcessFailureCount(getJobName()), not(0));
     }
 }
diff --git a/...taflow/throughput/StreamingThroughputDataFlowElasticJobForExecuteThrowsExceptionTest.java b/...taflow/throughput/StreamingThroughputDataFlowElasticJobForExecuteThrowsExceptionTest.java
@@ -18,6 +18,7 @@
 package com.dangdang.ddframe.job.integrate.std.dataflow.throughput;
 
 import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.not;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
@@ -49,6 +50,6 @@ public void assertJobInit() {
         }
         assertTrue(REG_CENTER.isExisted("/" + getJobName() + "/execution"));
         assertThat(ProcessCountStatistics.getProcessSuccessCount(getJobName()), is(0));
-        assertThat(ProcessCountStatistics.getProcessFailureCount(getJobName()), is(1));
+        assertThat(ProcessCountStatistics.getProcessFailureCount(getJobName()), not(0));
     }
 }
diff --git a/...-spring/src/test/java/com/dangdang/ddframe/job/spring/failure/ClassNotElasticJobTest.java b/...-spring/src/test/java/com/dangdang/ddframe/job/spring/failure/ClassNotElasticJobTest.java
@@ -31,8 +31,9 @@ public void setUp() {
         NestedZookeeperServers.getInstance().startServerIfNotStarted();
     }
 
+    @SuppressWarnings("resource")
     @Test(expected = BeanCreationException.class)
-    public void assertJobClassNotFound() {
+    public void assertNotElasticJob() {
         try {
             new ClassPathXmlApplicationContext("classpath:META-INF/job/classNotElasticJob.xml");
         } catch (final BeanCreationException ex) {

diff --git a/...c-job-spring/src/test/java/com/dangdang/ddframe/job/spring/failure/ClassNotFoundTest.java b/...c-job-spring/src/test/java/com/dangdang/ddframe/job/spring/failure/ClassNotFoundTest.java
@@ -31,6 +31,7 @@ public void setUp() {
         NestedZookeeperServers.getInstance().startServerIfNotStarted();
     }
 
+    @SuppressWarnings("resource")
     @Test(expected = BeanCreationException.class)
     public void assertJobClassNotFound() {
         try {