BATCHEE-112 persistence service doesnt support multiple execution for…

… getRunningExecution case of JobOperator when set to memory (default)

jbatch/src/main/java/org/apache/batchee/container/impl/JobOperatorImpl.java

@@ -255,7 +255,7 @@ public List<Long> getRunningExecutions(final String jobName) throws NoSuchJobExc
         // get the jobexecution ids associated with this job name
         final Set<Long> executionIds = persistenceManagerService.jobOperatorGetRunningExecutions(jobName);
 
-        if (executionIds.size() <= 0) {
+        if (executionIds.isEmpty()) {
             throw new NoSuchJobException("Job Name " + jobName + " not found");
         }
 

jbatch/src/main/java/org/apache/batchee/container/services/callback/SimpleJobExecutionCallbackService.java

@@ -57,7 +57,7 @@ public void waitFor(final long id) {
 
         // check before blocking
         final InternalJobExecution finalCheckExec = ServicesManager.find().service(BatchKernelService.class).getJobExecution(id);
-        if (Batches.isDone(finalCheckExec.getBatchStatus())) {
+        if (finalCheckExec != null && Batches.isDone(finalCheckExec.getBatchStatus())) {
             waiters.remove(id);
             return;
         }

jbatch/src/main/java/org/apache/batchee/container/services/persistence/MemoryPersistenceManagerService.java

@@ -232,7 +232,7 @@ public Properties getParameters(final long executionId) throws NoSuchJobExecutio
     public List<InternalJobExecution> jobOperatorGetJobExecutions(final long jobInstanceId) {
         final List<InternalJobExecution> list = new LinkedList<InternalJobExecution>();
         final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(jobInstanceId);
-        if (jobInstanceData == null || jobInstanceData.executions == null) {
+        if (jobInstanceData == null) {
             return list;
         }
         synchronized (jobInstanceData.executions) {
@@ -245,27 +245,20 @@ public List<InternalJobExecution> jobOperatorGetJobExecutions(final long jobInst
 
     @Override
     public Set<Long> jobOperatorGetRunningExecutions(final String jobName) {
-        Structures.JobInstanceData jobInstanceData = null;
-        for (final Structures.JobInstanceData instanceData : data.jobInstanceData.values()) {
-            if (instanceData.instance.getJobName().equals(jobName)) {
-                jobInstanceData = instanceData;
-                break;
-            }
-        }
-
-        if (jobInstanceData == null) {
-            return Collections.emptySet();
-        }
-
         final Set<Long> set = new HashSet<Long>();
 
-        synchronized (jobInstanceData.executions) {
-            for (final Structures.ExecutionInstanceData executionInstanceData : jobInstanceData.executions) {
-                if (RUNNING_STATUSES.contains(executionInstanceData.execution.getBatchStatus())) {
-                    set.add(executionInstanceData.execution.getExecutionId());
+        for (final Structures.JobInstanceData instanceData : data.jobInstanceData.values()) {
+            if (instanceData.instance.getJobName().equals(jobName)) {
+                synchronized (instanceData.executions) {
+                    for (final Structures.ExecutionInstanceData executionInstanceData : instanceData.executions) {
+                        if (RUNNING_STATUSES.contains(executionInstanceData.execution.getBatchStatus())) {
+                            set.add(executionInstanceData.execution.getExecutionId());
+                        }
+                    }
                 }
             }
         }
+
         return set;
     }
 

jbatch/src/test/java/org/apache/batchee/component/SimpleBatchlet.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2012 International Business Machines Corp.
+ *
+ * See the NOTICE file distributed with this work for additional information
+ * regarding copyright ownership. Licensed 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.component;
+
+import javax.batch.api.AbstractBatchlet;
+import javax.batch.api.BatchProperty;
+import javax.inject.Inject;
+
+import static java.lang.Thread.sleep;
+
+public class SimpleBatchlet extends AbstractBatchlet {
+    @Inject
+    @BatchProperty
+    private Long duration;
+
+    @Override
+    public String process() throws Exception {
+        if (duration != null && duration > 0) {
+            sleep(duration);
+        }
+        return "ok";
+    }
+}

jbatch/src/test/java/org/apache/batchee/container/impl/JobOperatorImplTest.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2012 International Business Machines Corp.
+ *
+ * See the NOTICE file distributed with this work for additional information
+ * regarding copyright ownership. Licensed 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.impl;
+
+import org.apache.batchee.container.services.ServicesManager;
+import org.apache.batchee.container.services.persistence.MemoryPersistenceManagerService;
+import org.apache.batchee.spi.PersistenceManagerService;
+import org.junit.Test;
+
+import javax.batch.operations.JobOperator;
+import java.util.List;
+import java.util.Properties;
+
+import static java.util.Collections.singletonList;
+import static org.apache.batchee.util.Batches.waitForEnd;
+import static org.junit.Assert.assertEquals;
+
+public class JobOperatorImplTest {
+    @Test
+    public void runningExecutionMemory_BATCHEE112() {
+        final JobOperator operator = new JobOperatorImpl(new ServicesManager() {{
+            init(new Properties() {{
+                setProperty(PersistenceManagerService.class.getSimpleName(), MemoryPersistenceManagerService.class.getName());
+            }});
+        }});
+        for (int i = 0; i < 10; i++) {
+            final long id = operator.start("simple", new Properties() {{
+                setProperty("duration", "3000");
+            }});
+            final List<Long> running = operator.getRunningExecutions("simple");
+            assertEquals("Iteration: " + i, singletonList(id), running);
+            waitForEnd(operator, id);
+        }
+    }
+}

jbatch/src/test/resources/META-INF/batch-jobs/simple.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- See the NOTICE file distributed with this work for additional information
+	regarding copyright ownership. Licensed 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. -->
+<job id="simple" version="1.0" xmlns="http://xmlns.jcp.org/xml/ns/javaee">
+  <step id="exec">
+    <batchlet ref="org.apache.batchee.component.SimpleBatchlet" />
+  </step>
+</job>