# How to execute activities in batch fashion in Temporal

**URL:** <https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349>\
**Category:** Community Support\
**Tags:** java-sdk\
**Created:** [June 10, 2021, 12:26pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349 "2021-06-10T12:26:34Z")\
**Posts on this page:** 20\
**Page:** 1

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 10, 2021, 12:26pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/1 "2021-06-10T12:26:34Z")

</div>

As mentioned in this [thread](https://community.temporal.io/t/limit-concurrent-activities-in-parallel/2066/5), I wanted to run 10 activities, but 4 activities asynchronously at time T.

As you responded there I’ll be starting with 4 activities and will be creating more if any activity gets completed.  
How will I achieve it via java code? Do I need to use `Promise.anyOf` to know if any activity gets completed. If so, if any one of the activities gets completed my workflow code gets completed.

If I try to add a newly created promise object in the promiseList then it is throwing `concurrentModificationException`

I tried the below code,

```auto
int limit = 4, size = 10;

for(int i=0; i<limit; i++){
    promiseList.add(Async.function(this::executeBindTask, task, taskIOIterator.next()));
}

    Promise.anyOf(promiseList).get();

List<Result> resultList = new ArrayList<>();
    for (Promise<Result> promise : promiseList) {
    if(promise.isCompleted()){
        resultList.add(promise.get());
        if(limit < size) {
            limit++;
            Async.function(this::executeBindTask, task, taskIOIterator.next()); // calling activity if any activity completes
        }
    }
}

```

Kindly help me to achieve it via java code. Thanks

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [June 10, 2021, 7:21pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/2 "2021-06-10T19:21:52Z")

</div>

Make sure that your main workflow method doesn’t return before all the computation is done. The simplest way to block the thread is to use `Workflow.await` on some variable like count of unprocessed items.

I doubt your code is going to work as I believe you cannot add to the list while iterating through it.

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 10, 2021, 7:37pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/3 "2021-06-10T19:37:17Z")

</div>

Yes, I did not update my list while iterating because it s throwing concurrentModification exception

```auto
if(limit < size) {
            limit++;
            Async.function(this::executeBindTask, task, taskIOIterator.next()); // calling activity if any activity completes
        }

```

In this code, I am not storing the promise object to some variable, but I have to store it into my promise list so that it will be included in `Promise.anyOf`.

like below,

```auto
 if(limit < size) {
            limit++;
            promiseList.add ( Async.function(this::executeBindTask, task, taskIOIterator.next()) ); // calling activity if any activity completes
        }

```

If I did not add my promise object in the list then how can put `.get` on the object?

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 10, 2021, 7:40pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/4 "2021-06-10T19:40:25Z")

</div>

Actually my use case is simple,

I wanted to execute 20 activities, but 4 tasks at a time. After executing 20 activities I have to collect all its results. That’s it.

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [June 10, 2021, 7:46pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/5 "2021-06-10T19:46:00Z")

</div>

I think you can simplify your code by using `Promise.thenApply` or `Promise.handle` callbacks.

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 10, 2021, 7:58pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/6 "2021-06-10T19:58:19Z")

</div>

Okay, let me try out

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 11, 2021, 5:47am UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/7 "2021-06-11T05:47:15Z")

</div>

Thanks @maxim it is working, but not for few cases as my workflow code is not deterministic.

Q1). In the java doc, it says that `(Workflow.await(condition))``condition is called on every state transition,` what does state transition mean?

As you said I have a variable called `count` which is reduced for every `promise callbacks promise.thenApply`

I have the below await call to block my code. count initialized with listSize;  
`Workflow.await(()-> count == 0);`

If the worker has restarted in the middle of the execution then my count is re-initialized with listSize and it never reaches zero though all my activities are completed.

Q2). What change should I do to make my code to be deterministic?

**Note: If the worker has not restarted then it is executing as expected**

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 11, 2021, 6:01am UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/8 "2021-06-11T06:01:52Z")

</div>

@maxim `Workflow.await` will block the workflow thread similar to `Promise.allof(promiseList)`.

Am I correct?

---

<div class="post-metadata">

**Author:** ![tihomir](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/tihomir/32/6580_2.png) [@tihomir](https://community.temporal.io/u/tihomir)\
**Post date:** [June 11, 2021, 2:27pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/9 "2021-06-11T14:27:26Z")

</div>

Workflow.await blocks the current thread until the unblocking condition you specify is evaluated to true.

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 11, 2021, 2:43pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/10 "2021-06-11T14:43:30Z")

</div>

Okay, Promise.allOf also does the same thing right? it will block the current thread until all of its promise objects return the results.

---

<div class="post-metadata">

**Author:** ![tihomir](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/tihomir/32/6580_2.png) [@tihomir](https://community.temporal.io/u/tihomir)\
**Post date:** [June 11, 2021, 2:56pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/11 "2021-06-11T14:56:01Z")

</div>

You block/wait on the `.get()` I believe.

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 11, 2021, 3:12pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/12 "2021-06-11T15:12:12Z")

</div>

Both aren’t the same 😧

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [June 11, 2021, 3:36pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/13 "2021-06-11T15:36:10Z")

</div>

`Promise.allOf` returns a promise that you want to block on by calling its method get.

```java
Promise.allOf(list).get();

```

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 11, 2021, 4:13pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/14 "2021-06-11T16:13:43Z")

</div>

My argument is `Workflow.await(cond)` and `Promise.allOf(list).get();`  
Q1). Both block the current workflow thread, right?

if not kindly let me know the differences?

Q2).

> [@anil\_kumble](#):
>
> Q1). In the java doc, it says that `(Workflow.await(condition))` `condition is called on every state transition,` what does state transition mean?
> 
> As you said I have a variable called `count` which is reduced for every `promise callbacks promise.thenApply`
> 
> I have the below await call to block my code. count initialized with listSize;  
> `Workflow.await(()-> count == 0);`
> 
> If the worker has restarted in the middle of the execution then my count is re-initialized with listSize and it never reaches zero though all my activities are completed.
> 
> Q2). What change should I do to make my code to be deterministic?
> 
> **Note: If the worker has not restarted then it is executing as expected**

Please give you insight regarding this

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [June 11, 2021, 4:19pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/15 "2021-06-11T16:19:16Z")

</div>

> if not kindly let me know the differences?

await blocks until the condition is satisfied. `Promise.allOf(list).get` blocks until all the promises in the list are ready.

> Please give you insight regarding this

The count should not be reset to 0 on worker restart. Could you post the code that defines and uses the counter?

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 11, 2021, 4:34pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/16 "2021-06-11T16:34:07Z")

</div>

> [@maxim](#):
>
> The count should not be reset to 0 on worker restart. Could you post the code that defines and uses the counter?

> Assume below value,  
> batchTask.getIoCollection().size() = 20. (size = 20)  
> windowSize = 4;

I will be adding my results into resultList, once resultList reaches the size, await will be released.  
initially, it is declared with an empty list

Before I had a variable `count`, on each callback I will be increasing it and using the same in await also. Since I need the resultList, I removed the counter and used my resultList for checking.

```auto
private Result executeBatchTask(BatchTask batchTask, TaskIO taskIO) {

        Task task = batchTask.getTask();
        AtomicInteger index = new AtomicInteger(0);
        Integer size = batchTask.getIoCollection().size();
        List<Result> resultList = Collections.synchronizedList(new ArrayList<Result>());

        Iterator<TaskIO> taskIOIterator = batchTask.getIoCollection().iterator();

        for(; index.get()<batchTask.getWindowSize(); index.incrementAndGet()){
            this.runBindTaskTask(task, taskIOIterator, index, size, resultList);
        }

        Workflow.await(()-> resultList.size() == size);

        return batchTask.handleResults(taskIO, resultList);
    }

    private void runBindTaskTask(Task task, Iterator<TaskIO> taskIOIterator, AtomicInteger index, Integer size, List<Result> resultList) {

        Promise<Result> promise = Async.function(this::runBindTaskTask, task, taskIOIterator.next());

        promise.thenApply(result->{
            resultList.add(result);
            if(index.get() != size){
                index.incrementAndGet();
                this.runBindTaskTask(task, taskIOIterator, index, size, resultList);
            }
            return result;
        });
    }

    private Result runBindTaskTask(Task task, TaskIO taskIO){
       
/* activity call */
        taskIO = runnableTask.run((FunctionalTask) task, taskIO); 
        return Result.getSuccessResult(taskIO);
    }

```

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 11, 2021, 8:13pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/17 "2021-06-11T20:13:16Z")

</div>

@maxim Can you please give your suggestion regarding this

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [June 15, 2021, 2:22pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/18 "2021-06-15T14:22:48Z")

</div>

According to [documentation](https://docs.temporal.io/docs/java/workflows#workflow-implementation-constraints) using Java synchronization is not allowed in the workflow code. Don’t use `synchronizedList` inside the workflow code as it breaks this rule.

I think you might have but around index as you check it for boundary condition first and then increment in the `thenApply` method.

In general the code should work after all the bugs are fixed.

---

<div class="post-metadata">

**Author:** ![anil\_kumble](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/anil_kumble/32/954_2.png) [@anil\_kumble](https://community.temporal.io/u/anil_kumble)\
**Post date:** [June 15, 2021, 4:37pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/19 "2021-06-15T16:37:31Z")

</div>

> [@maxim](#):
>
> According to [documentation](https://docs.temporal.io/docs/java/workflows?&_ga=2.248296512.657997049.1623260525-606638953.1620997252#workflow-implementation-constraints) using Java synchronization is not allowed in the workflow code. Don’t use `synchronizedList` inside the workflow code as it breaks this rule.

Okay, let me remove `synchronizedList `. Can I use `AtomicInteger` ?

> [@maxim](#):
>
> I think you might have but around index as you check it for boundary condition first and then increment in the `thenApply` method.

Sorry , I did not understand what you are saying

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [June 15, 2021, 11:22pm UTC](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349/20 "2021-06-15T23:22:42Z")

</div>

I might be wrong, but it looks like that you first call an activity with index == size and then check for it.

[Next page](https://community.temporal.io/t/how-to-execute-activities-in-batch-fashion-in-temporal/2349.md?page=2)