You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@batchee.apache.org by rm...@apache.org on 2014/01/08 08:07:52 UTC
git commit: BATCHEE-14 adding JobExecutionCallbackService
Updated Branches:
refs/heads/master 8688a28d8 -> da630a696
BATCHEE-14 adding JobExecutionCallbackService
Project: http://git-wip-us.apache.org/repos/asf/incubator-batchee/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-batchee/commit/da630a69
Tree: http://git-wip-us.apache.org/repos/asf/incubator-batchee/tree/da630a69
Diff: http://git-wip-us.apache.org/repos/asf/incubator-batchee/diff/da630a69
Branch: refs/heads/master
Commit: da630a696c6925430b01baedac066cfc6ec8d598
Parents: 8688a28
Author: Romain Manni-Bucau <rm...@apache.org>
Authored: Wed Jan 8 08:07:54 2014 +0100
Committer: Romain Manni-Bucau <rm...@apache.org>
Committed: Wed Jan 8 08:07:54 2014 +0100
----------------------------------------------------------------------
.../batchee/container/impl/JobOperatorImpl.java | 7 ++
.../container/services/ServicesManager.java | 3 +
.../SimpleJobExecutionCallbackService.java | 78 ++++++++++++++++++++
.../services/kernel/DefaultBatchKernel.java | 9 ++-
.../spi/JobExecutionCallbackService.java | 24 ++++++
.../java/org/apache/batchee/util/Batches.java | 12 +++
.../batchee/test/SynchronousJobOperator.java | 11 +++
7 files changed, 142 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-batchee/blob/da630a69/jbatch/src/main/java/org/apache/batchee/container/impl/JobOperatorImpl.java
----------------------------------------------------------------------
diff --git a/jbatch/src/main/java/org/apache/batchee/container/impl/JobOperatorImpl.java b/jbatch/src/main/java/org/apache/batchee/container/impl/JobOperatorImpl.java
index 28fab55..1425f02 100755
--- a/jbatch/src/main/java/org/apache/batchee/container/impl/JobOperatorImpl.java
+++ b/jbatch/src/main/java/org/apache/batchee/container/impl/JobOperatorImpl.java
@@ -25,6 +25,7 @@ import org.apache.batchee.container.status.JobStatus;
import org.apache.batchee.jmx.BatchEE;
import org.apache.batchee.jmx.BatchEEMBean;
import org.apache.batchee.jmx.BatchEEMBeanImpl;
+import org.apache.batchee.spi.JobExecutionCallbackService;
import org.apache.batchee.spi.JobXMLLoaderService;
import org.apache.batchee.spi.PersistenceManagerService;
import org.apache.batchee.spi.SecurityService;
@@ -106,6 +107,7 @@ public class JobOperatorImpl implements JobOperator {
private final JobXMLLoaderService xmlLoaderService;
private final JobStatusManagerService statusManagerService;
private final SecurityService securityService;
+ private final JobExecutionCallbackService callbackService;
public JobOperatorImpl() {
final ServicesManager servicesManager = ServicesManager.find();
@@ -114,6 +116,7 @@ public class JobOperatorImpl implements JobOperator {
xmlLoaderService = servicesManager.service(JobXMLLoaderService.class);
statusManagerService = servicesManager.service(JobStatusManagerService.class);
securityService = servicesManager.service(SecurityService.class);
+ callbackService = servicesManager.service(JobExecutionCallbackService.class);
}
@Override
@@ -380,4 +383,8 @@ public class JobOperatorImpl implements JobOperator {
kernelService.stopJob(executionId);
}
+
+ public void waitFor(final long id) {
+ callbackService.waitFor(id);
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-batchee/blob/da630a69/jbatch/src/main/java/org/apache/batchee/container/services/ServicesManager.java
----------------------------------------------------------------------
diff --git a/jbatch/src/main/java/org/apache/batchee/container/services/ServicesManager.java b/jbatch/src/main/java/org/apache/batchee/container/services/ServicesManager.java
index 892ae61..141cc3f 100755
--- a/jbatch/src/main/java/org/apache/batchee/container/services/ServicesManager.java
+++ b/jbatch/src/main/java/org/apache/batchee/container/services/ServicesManager.java
@@ -19,6 +19,7 @@ package org.apache.batchee.container.services;
import org.apache.batchee.container.exception.BatchContainerRuntimeException;
import org.apache.batchee.container.exception.BatchContainerServiceException;
+import org.apache.batchee.container.services.callback.SimpleJobExecutionCallbackService;
import org.apache.batchee.container.services.executor.DefaultThreadPoolService;
import org.apache.batchee.container.services.factory.CDIBatchArtifactFactory;
import org.apache.batchee.container.services.factory.DefaultBatchArtifactFactory;
@@ -33,6 +34,7 @@ import org.apache.batchee.container.util.BatchContainerConstants;
import org.apache.batchee.spi.BatchArtifactFactory;
import org.apache.batchee.spi.BatchService;
import org.apache.batchee.spi.BatchThreadPoolService;
+import org.apache.batchee.spi.JobExecutionCallbackService;
import org.apache.batchee.spi.JobXMLLoaderService;
import org.apache.batchee.spi.PersistenceManagerService;
import org.apache.batchee.spi.SecurityService;
@@ -62,6 +64,7 @@ public class ServicesManager implements BatchContainerConstants {
SERVICE_IMPL_CLASS_NAMES.put(BatchKernelService.class.getName(), DefaultBatchKernel.class.getName());
SERVICE_IMPL_CLASS_NAMES.put(JobXMLLoaderService.class.getName(), DefaultJobXMLLoaderService.class.getName());
SERVICE_IMPL_CLASS_NAMES.put(SecurityService.class.getName(), DefaultSecurityService.class.getName());
+ SERVICE_IMPL_CLASS_NAMES.put(JobExecutionCallbackService.class.getName(), SimpleJobExecutionCallbackService.class.getName());
try {
Thread.currentThread().getContextClassLoader().loadClass("javax.enterprise.inject.spi.BeanManager");
SERVICE_IMPL_CLASS_NAMES.put(BatchArtifactFactory.class.getName(), CDIBatchArtifactFactory.class.getName());
http://git-wip-us.apache.org/repos/asf/incubator-batchee/blob/da630a69/jbatch/src/main/java/org/apache/batchee/container/services/callback/SimpleJobExecutionCallbackService.java
----------------------------------------------------------------------
diff --git a/jbatch/src/main/java/org/apache/batchee/container/services/callback/SimpleJobExecutionCallbackService.java b/jbatch/src/main/java/org/apache/batchee/container/services/callback/SimpleJobExecutionCallbackService.java
new file mode 100644
index 0000000..1416894
--- /dev/null
+++ b/jbatch/src/main/java/org/apache/batchee/container/services/callback/SimpleJobExecutionCallbackService.java
@@ -0,0 +1,78 @@
+/*
+ * 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.batchee.container.services.callback;
+
+import org.apache.batchee.container.exception.BatchContainerRuntimeException;
+import org.apache.batchee.container.impl.jobinstance.RuntimeJobExecution;
+import org.apache.batchee.container.services.BatchKernelService;
+import org.apache.batchee.container.services.InternalJobExecution;
+import org.apache.batchee.container.services.ServicesManager;
+import org.apache.batchee.spi.JobExecutionCallbackService;
+import org.apache.batchee.util.Batches;
+
+import java.util.Collection;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+
+public class SimpleJobExecutionCallbackService implements JobExecutionCallbackService {
+ private final ConcurrentMap<Long, Collection<CountDownLatch>> waiters = new ConcurrentHashMap<Long, Collection<CountDownLatch>>();
+
+ @Override
+ public void onJobExecutionDone(final RuntimeJobExecution jobExecution) {
+ final Collection<CountDownLatch> toRealease = waiters.remove(jobExecution.getExecutionId());
+ if (toRealease != null) {
+ for (final CountDownLatch latch : toRealease) {
+ latch.countDown();
+ }
+ }
+ }
+
+ @Override
+ public void waitFor(final long id) {
+ Collection<CountDownLatch> toRealease = waiters.remove(id);
+ if (toRealease == null) {
+ toRealease = new CopyOnWriteArrayList<CountDownLatch>();
+ final Collection<CountDownLatch> existing = waiters.putIfAbsent(id, toRealease);
+ if (existing != null) {
+ toRealease = existing;
+ }
+ }
+
+ // check before blocking
+ final InternalJobExecution finalCheckExec = ServicesManager.find().service(BatchKernelService.class).getJobExecution(id);
+ if (Batches.isDone(finalCheckExec.getBatchStatus())) {
+ waiters.remove(id);
+ return;
+ }
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ toRealease.add(latch);
+ try {
+ latch.await();
+ } catch (final InterruptedException e) {
+ throw new BatchContainerRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void init(final Properties batchConfig) {
+ // no-op
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-batchee/blob/da630a69/jbatch/src/main/java/org/apache/batchee/container/services/kernel/DefaultBatchKernel.java
----------------------------------------------------------------------
diff --git a/jbatch/src/main/java/org/apache/batchee/container/services/kernel/DefaultBatchKernel.java b/jbatch/src/main/java/org/apache/batchee/container/services/kernel/DefaultBatchKernel.java
index 7dc488a..f69ad95 100755
--- a/jbatch/src/main/java/org/apache/batchee/container/services/kernel/DefaultBatchKernel.java
+++ b/jbatch/src/main/java/org/apache/batchee/container/services/kernel/DefaultBatchKernel.java
@@ -32,6 +32,7 @@ import org.apache.batchee.container.util.FlowInSplitBuilderConfig;
import org.apache.batchee.container.util.PartitionsBuilderConfig;
import org.apache.batchee.jaxb.JSLJob;
import org.apache.batchee.spi.BatchThreadPoolService;
+import org.apache.batchee.spi.JobExecutionCallbackService;
import org.apache.batchee.spi.PersistenceManagerService;
import javax.batch.operations.JobExecutionAlreadyCompleteException;
@@ -58,11 +59,13 @@ public class DefaultBatchKernel implements BatchKernelService {
private final BatchThreadPoolService executorService;
private final PersistenceManagerService persistenceService;
private final ServicesManager servicesManager;
+ private final JobExecutionCallbackService jobExecutionCallback;
public DefaultBatchKernel(final ServicesManager servicesManager) {
this.servicesManager = servicesManager;
- executorService = servicesManager.service(BatchThreadPoolService.class);
- persistenceService = servicesManager.service(PersistenceManagerService.class);
+ this.executorService = servicesManager.service(BatchThreadPoolService.class);
+ this.persistenceService = servicesManager.service(PersistenceManagerService.class);
+ this.jobExecutionCallback = servicesManager.service(JobExecutionCallbackService.class);
}
@Override
@@ -120,6 +123,8 @@ public class DefaultBatchKernel implements BatchKernelService {
}
}
+ jobExecutionCallback.onJobExecutionDone(jobExecution);
+
// AJM: ah - purge jobExecution from map here and flush to DB?
// edit: no long want a 2 tier for the jobexecution...do want it for step execution
// renamed method to flushAndRemoveStepExecution
http://git-wip-us.apache.org/repos/asf/incubator-batchee/blob/da630a69/jbatch/src/main/java/org/apache/batchee/spi/JobExecutionCallbackService.java
----------------------------------------------------------------------
diff --git a/jbatch/src/main/java/org/apache/batchee/spi/JobExecutionCallbackService.java b/jbatch/src/main/java/org/apache/batchee/spi/JobExecutionCallbackService.java
new file mode 100644
index 0000000..2496292
--- /dev/null
+++ b/jbatch/src/main/java/org/apache/batchee/spi/JobExecutionCallbackService.java
@@ -0,0 +1,24 @@
+/*
+ * 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.batchee.spi;
+
+import org.apache.batchee.container.impl.jobinstance.RuntimeJobExecution;
+
+public interface JobExecutionCallbackService extends BatchService {
+ void onJobExecutionDone(RuntimeJobExecution jobExecution);
+ void waitFor(long id);
+}
http://git-wip-us.apache.org/repos/asf/incubator-batchee/blob/da630a69/jbatch/src/main/java/org/apache/batchee/util/Batches.java
----------------------------------------------------------------------
diff --git a/jbatch/src/main/java/org/apache/batchee/util/Batches.java b/jbatch/src/main/java/org/apache/batchee/util/Batches.java
index b038a2e..a7668c7 100644
--- a/jbatch/src/main/java/org/apache/batchee/util/Batches.java
+++ b/jbatch/src/main/java/org/apache/batchee/util/Batches.java
@@ -16,6 +16,8 @@
*/
package org.apache.batchee.util;
+import org.apache.batchee.container.impl.JobOperatorImpl;
+
import javax.batch.operations.JobOperator;
import javax.batch.runtime.BatchRuntime;
import javax.batch.runtime.BatchStatus;
@@ -34,6 +36,12 @@ public class Batches {
}
public static void waitForEnd(final JobOperator jobOperator, final long id) {
+ if (JobOperatorImpl.class.isInstance(jobOperator)) {
+ JobOperatorImpl.class.cast(jobOperator).waitFor(id);
+ return;
+ }
+
+ // else polling
do {
try {
Thread.sleep(100);
@@ -43,6 +51,10 @@ public class Batches {
} while (!isDone(jobOperator, id));
}
+ public static boolean isDone(final BatchStatus status) {
+ return BATCH_END_STATUSES.contains(status);
+ }
+
public static boolean isDone(final JobOperator jobOperator, final long id) {
return BATCH_END_STATUSES.contains(jobOperator.getJobExecution(id).getBatchStatus());
}
http://git-wip-us.apache.org/repos/asf/incubator-batchee/blob/da630a69/test/src/main/java/org/apache/batchee/test/SynchronousJobOperator.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/batchee/test/SynchronousJobOperator.java b/test/src/main/java/org/apache/batchee/test/SynchronousJobOperator.java
index 8051b6f..6fa078b 100644
--- a/test/src/main/java/org/apache/batchee/test/SynchronousJobOperator.java
+++ b/test/src/main/java/org/apache/batchee/test/SynchronousJobOperator.java
@@ -61,6 +61,17 @@ public class SynchronousJobOperator implements JobOperator {
}
private void waitEnd(final long id) { // copy of Batches class but avoids to be linked to BatchEE
+ try {
+ final Class<?> batcheeOpClass = Thread.currentThread().getContextClassLoader().loadClass("org.apache.batchee.container.impl.JobOperatorImpl");
+ if (batcheeOpClass.isInstance(delegate)) {
+ batcheeOpClass.getMethod("waitFor", long.class).invoke(delegate, id);
+ return;
+ }
+ return;
+ } catch (final Exception e) {
+ // no-op
+ }
+
do {
try {
Thread.sleep(20);