-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[feature]implementing parallel task execution
PRD-202368
- Loading branch information
1 parent
34d5809
commit 6ca7821
Showing
15 changed files
with
443 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
95 changes: 95 additions & 0 deletions
95
...ng-boot-starter/src/main/java/com/tosan/camunda/camundaclient/config/ExecutionConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| package com.tosan.camunda.camundaclient.config; | ||
|
|
||
| import com.tosan.camunda.camundaclient.config.enumuration.ExecutionType; | ||
|
|
||
| /** | ||
| * @author Shahryar Safizadeh | ||
| * @since 12/4/2023 | ||
| */ | ||
| public class ExecutionConfig { | ||
| /** | ||
| * wait duration per task in parallel backoff strategy | ||
| */ | ||
| private Long waitDurationPerTask; | ||
|
|
||
| /** | ||
| * thread pool size for parallel executor | ||
| */ | ||
| private int threadPoolSize; | ||
|
|
||
| /** | ||
| * type of execution | ||
| */ | ||
| private ExecutionType executionType; | ||
|
|
||
| /** | ||
| * is multi-instance enabled | ||
| */ | ||
| private boolean isMultiInstanceEnabled; | ||
|
|
||
| /** | ||
| * @return wait duration per task in parallel backoff strategy | ||
| */ | ||
| public Long getWaitDurationPerTask() { | ||
| return waitDurationPerTask; | ||
| } | ||
|
|
||
| /** | ||
| * @param waitDurationPerTask wait duration per task in parallel backoff strategy | ||
| */ | ||
| public void setWaitDurationPerTask(Long waitDurationPerTask) { | ||
| this.waitDurationPerTask = waitDurationPerTask; | ||
| } | ||
|
|
||
| /** | ||
| * @return thread pool size for parallel executor | ||
| */ | ||
| public int getThreadPoolSize() { | ||
| return threadPoolSize; | ||
| } | ||
|
|
||
| /** | ||
| * @param threadPoolSize thread pool size for parallel executor | ||
| */ | ||
| public void setThreadPoolSize(int threadPoolSize) { | ||
| this.threadPoolSize = threadPoolSize; | ||
| } | ||
|
|
||
| /** | ||
| * @return type of execution | ||
| */ | ||
| public ExecutionType getExecutionType() { | ||
| return executionType; | ||
| } | ||
|
|
||
| /** | ||
| * @param executionType type of execution | ||
| */ | ||
| public void setExecutionType(ExecutionType executionType) { | ||
| this.executionType = executionType; | ||
| } | ||
|
|
||
| /** | ||
| * @return is multi-instance enabled | ||
| */ | ||
| public boolean isMultiInstanceEnabled() { | ||
| return isMultiInstanceEnabled; | ||
| } | ||
|
|
||
| /** | ||
| * @param multiInstanceEnabled is multi-instance enabled | ||
| */ | ||
| public void setMultiInstanceEnabled(boolean multiInstanceEnabled) { | ||
| isMultiInstanceEnabled = multiInstanceEnabled; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "ExecutionConfig{" + | ||
| "waitDurationPerTask=" + waitDurationPerTask + | ||
| ", threadPoolSize=" + threadPoolSize + | ||
| ", executionType=" + executionType + | ||
| ", isMultiInstanceEnabled=" + isMultiInstanceEnabled + | ||
| '}'; | ||
| } | ||
| } |
76 changes: 76 additions & 0 deletions
76
...starter/src/main/java/com/tosan/camunda/camundaclient/config/ParallelBackoffStrategy.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| package com.tosan.camunda.camundaclient.config; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.camunda.bpm.client.backoff.ExponentialBackoffStrategy; | ||
| import org.camunda.bpm.client.task.ExternalTask; | ||
| import org.springframework.beans.factory.annotation.Value; | ||
| import org.springframework.validation.annotation.Validated; | ||
|
|
||
| import java.util.Date; | ||
| import java.util.List; | ||
| import java.util.concurrent.Future; | ||
|
|
||
| /** | ||
| * @author Shahryar Safizadeh | ||
| * @since 11/20/2023 | ||
| */ | ||
| @Slf4j | ||
| @Validated | ||
| public class ParallelBackoffStrategy extends ExponentialBackoffStrategy { | ||
|
|
||
| @Value("${camunda.bpm.client.execution.wait-duration-per-task}") | ||
| private Long waitDurationPerTask; | ||
|
|
||
| private ParallelTaskExecutor parallelTaskExecutor; | ||
|
|
||
| public ParallelBackoffStrategy(ParallelTaskExecutor parallelTaskExecutor) { | ||
| this.parallelTaskExecutor = parallelTaskExecutor; | ||
| } | ||
|
|
||
| @Override | ||
| public void reconfigure(List<ExternalTask> externalTasks) { | ||
| super.reconfigure(externalTasks); | ||
| if (!externalTasks.isEmpty()) { | ||
| try { | ||
| waitForTaskCompletion(externalTasks); | ||
| } catch (InterruptedException | RuntimeException e) { | ||
| log.error("Error happened in main thread while waiting for tasks in backoff strategy!!!", e); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private void waitForTaskCompletion(List<ExternalTask> externalTasks) throws InterruptedException { | ||
| if (waitDurationPerTask == null) { | ||
| throw new RuntimeException("camunda.bpm.client.execution.wait-duration-per-task is not set or is not valid."); | ||
| } else { | ||
| Date expireDate = new Date(System.currentTimeMillis() + waitDurationPerTask * externalTasks.size()); | ||
| while (!isAllTasksCompleted(parallelTaskExecutor.getFutures())) { | ||
| Date currentTime = new Date(System.currentTimeMillis()); | ||
| if (expireDate.before(currentTime)) { | ||
| break; | ||
| } | ||
| Thread.sleep(5000); | ||
| } | ||
| cancelAllFutures(parallelTaskExecutor.getFutures()); | ||
| parallelTaskExecutor.getFutures().clear(); | ||
| } | ||
| } | ||
|
|
||
| private void cancelAllFutures(List<Future<?>> futures) { | ||
| for (Future future : futures) { | ||
| if (!future.isDone() && !future.isCancelled()) { | ||
| log.error("Future : {} did not complete before wait duration time and got canceled.", future); | ||
| future.cancel(true); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private boolean isAllTasksCompleted(List<Future<?>> futures) { | ||
| for (Future future : futures) { | ||
| if (!future.isDone()) { | ||
| return false; | ||
| } | ||
| } | ||
| return true; | ||
| } | ||
| } |
56 changes: 56 additions & 0 deletions
56
...ot-starter/src/main/java/com/tosan/camunda/camundaclient/config/ParallelTaskExecutor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| package com.tosan.camunda.camundaclient.config; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.aspectj.lang.ProceedingJoinPoint; | ||
| import org.camunda.bpm.client.task.ExternalTask; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.concurrent.*; | ||
|
|
||
| /** | ||
| * @author Shahryar Safizadeh | ||
| * @since 11/20/2023 | ||
| */ | ||
| @Slf4j | ||
| public class ParallelTaskExecutor { | ||
| private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1); | ||
| private ExecutorService executorService; | ||
| private static List<Future<?>> futures = new ArrayList<>(); | ||
|
|
||
| public ParallelTaskExecutor(ExecutorService executorService) { | ||
| this.executorService = executorService; | ||
| } | ||
|
|
||
| public Future<?> submitToExecutor(ProceedingJoinPoint pjp) { | ||
| Future<?> future = executorService.submit(() -> { | ||
| try { | ||
| pjp.proceed(); | ||
| } catch (Throwable e) { | ||
| if (e instanceof RuntimeException) { | ||
| RuntimeException runtimeException = (RuntimeException) e; | ||
| throw runtimeException; | ||
| } | ||
| throw new RuntimeException(e); | ||
| } | ||
| }); | ||
| return future; | ||
| } | ||
|
|
||
| public void scheduledTaskCancellation(Future<?> future, ExternalTask externalTask) { | ||
| scheduledExecutorService.schedule(() -> { | ||
| if (!future.isDone()) { | ||
| log.info("Task : {} is canceled by scheduledCanceller!!!", externalTask.getBusinessKey()); | ||
| future.cancel(true); | ||
| } | ||
| }, getLockExpirationTimeDuration(externalTask), TimeUnit.MILLISECONDS); | ||
| } | ||
|
|
||
| public long getLockExpirationTimeDuration(ExternalTask externalTask) { | ||
| return externalTask.getLockExpirationTime().getTime() - System.currentTimeMillis(); | ||
| } | ||
|
|
||
| public List<Future<?>> getFutures() { | ||
| return futures; | ||
| } | ||
| } |
Oops, something went wrong.