diff --git a/pom.xml b/pom.xml index 462198a..35f60d6 100644 --- a/pom.xml +++ b/pom.xml @@ -12,7 +12,7 @@ camunda8-adapter VanillaBP SPI adapter for Camunda 8.x - 1.3.1-SNAPSHOT + 1.4.0-SNAPSHOT pom diff --git a/spring-boot/README.md b/spring-boot/README.md index b2dfcf8..20ed3cc 100644 --- a/spring-boot/README.md +++ b/spring-boot/README.md @@ -190,15 +190,51 @@ Since Camunda 8 is an external system to your services one has to deal with even 1. Camunda 8 is not reachable / cannot process the confirmation of a completed task 1. The task to be completed was cancelled meanwhile e.g. due to boundary events -If there is an exception in your business code and you have to roll back the transaction then Camunda's task retry-mechanism should retry as configured. Additionally, the `TaskException` is used for expected business errors handled by BPMN error boundary events which must not cause a rollback. To achieve both one should mark the service bean like this: +In order to activate this behavior one has to mark methods accessing VanillaBP-APIs as `@Transactional`, either by +using the method-level annotation: ```java @Service @WorkflowService(workflowAggregateClass = Ride.class) -@Transactional(noRollbackFor = TaskException.class) public class TaxiRide { + @Autowired + private ProcessService processService; + + @Transactional + public void receivePayment(...) { + ... + processService.startWorkflow(ride); + ... + } + + @Transactional + @WorkflowTask + public void chargeCreditCard(final Ride ride) { + ... + } + + @Transactional + public void paymentReceived(final Ride ride) { + ... + processService.correlateMessage(ride, 'PaymentReceived'); + ... + } +} +``` + +or the class-level annotation: + +```java +@Service +@WorkflowService(workflowAggregateClass = Ride.class) +@Transactional +public class TaxiRide { + ... +} ``` +If there is an exception in your business code and you have to roll back the transaction then Camunda's task retry-mechanism should retry as configured. Additionally, the `TaskException` is used for expected business errors handled by BPMN error boundary events which must not cause a rollback. This is handled by the adapter, one does not need to take care about it. + ## Workflow aggregate serialization On using C7 one can use workflow aggregates having relations and calculated values: diff --git a/spring-boot/pom.xml b/spring-boot/pom.xml index 94d4aa6..a990186 100644 --- a/spring-boot/pom.xml +++ b/spring-boot/pom.xml @@ -5,7 +5,7 @@ org.camunda.community.vanillabp camunda8-adapter - 1.3.1-SNAPSHOT + 1.4.0-SNAPSHOT camunda8-spring-boot-adapter @@ -13,7 +13,7 @@ UTF-8 - 8.3.4.2 + 8.5.4 @@ -38,14 +38,19 @@ 1.1.1 - io.camunda - spring-zeebe-starter + io.camunda.spring + spring-boot-starter-camunda ${spring.zeebe.version} org.springframework spring-tx - 5.3.23 + 6.1.3 + + + org.springframework.boot + spring-boot-starter-aop + 3.2.5 jakarta.persistence diff --git a/spring-boot/src/main/java/io/vanillabp/camunda8/Camunda8AdapterConfiguration.java b/spring-boot/src/main/java/io/vanillabp/camunda8/Camunda8AdapterConfiguration.java index 43b280c..c947d8d 100644 --- a/spring-boot/src/main/java/io/vanillabp/camunda8/Camunda8AdapterConfiguration.java +++ b/spring-boot/src/main/java/io/vanillabp/camunda8/Camunda8AdapterConfiguration.java @@ -1,12 +1,13 @@ package io.vanillabp.camunda8; import io.camunda.zeebe.spring.client.CamundaAutoConfiguration; -import io.camunda.zeebe.spring.client.jobhandling.DefaultCommandExceptionHandlingStrategy; import io.vanillabp.camunda8.deployment.Camunda8DeploymentAdapter; import io.vanillabp.camunda8.deployment.DeploymentRepository; import io.vanillabp.camunda8.deployment.DeploymentResourceRepository; import io.vanillabp.camunda8.deployment.DeploymentService; import io.vanillabp.camunda8.service.Camunda8ProcessService; +import io.vanillabp.camunda8.service.Camunda8TransactionAspect; +import io.vanillabp.camunda8.service.Camunda8TransactionProcessor; import io.vanillabp.camunda8.wiring.Camunda8Connectable.Type; import io.vanillabp.camunda8.wiring.Camunda8TaskHandler; import io.vanillabp.camunda8.wiring.Camunda8TaskWiring; @@ -17,6 +18,8 @@ import io.vanillabp.springboot.adapter.VanillaBpProperties; import io.vanillabp.springboot.parameters.MethodParameter; import jakarta.annotation.PostConstruct; +import java.lang.reflect.Method; +import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.aop.framework.AopProxyUtils; @@ -29,13 +32,13 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Scope; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; import org.springframework.data.repository.CrudRepository; -import java.lang.reflect.Method; -import java.util.List; - @AutoConfigurationPackage(basePackageClasses = Camunda8AdapterConfiguration.class) @AutoConfigureBefore(CamundaAutoConfiguration.class) @EnableConfigurationProperties(Camunda8VanillaBpProperties.class) @@ -57,9 +60,6 @@ public class Camunda8AdapterConfiguration extends AdapterConfigurationBase Camunda8ProcessService newProcessServiceImplementation( final var result = new Camunda8ProcessService( camunda8Properties, + eventPublisher, workflowAggregateRepository, - workflowAggregate -> springDataUtil.getId(workflowAggregate), + springDataUtil::getId, workflowAggregateClass); putConnectableService(workflowAggregateClass, result); @@ -185,4 +196,31 @@ public SpringBeanUtil vanillabpSpringBeanUtil( } + /* + * https://www.tirasa.net/en/blog/dynamic-spring-s-transactional-2020-edition + */ + /* + @Bean + public static BeanFactoryPostProcessor camunda8TransactionInterceptorInjector() { + + return beanFactory -> { + String[] names = beanFactory.getBeanNamesForType(TransactionInterceptor.class); + for (String name : names) { + BeanDefinition bd = beanFactory.getBeanDefinition(name); + bd.setBeanClassName(Camunda8TransactionInterceptor.class.getName()); + bd.setFactoryBeanName(null); + bd.setFactoryMethodName(null); + } + }; + + } + */ + + @Bean + public Camunda8TransactionProcessor camunda8TransactionProcessor() { + + return new Camunda8TransactionProcessor(); + + } + } diff --git a/spring-boot/src/main/java/io/vanillabp/camunda8/Camunda8VanillaBpProperties.java b/spring-boot/src/main/java/io/vanillabp/camunda8/Camunda8VanillaBpProperties.java index 26e459b..e5b03de 100644 --- a/spring-boot/src/main/java/io/vanillabp/camunda8/Camunda8VanillaBpProperties.java +++ b/spring-boot/src/main/java/io/vanillabp/camunda8/Camunda8VanillaBpProperties.java @@ -39,8 +39,8 @@ public String getTenantId( if (!configuration.isUseTenants()) { return null; } - if (StringUtils.hasText(configuration.getTenant())) { - return configuration.getTenant(); + if (StringUtils.hasText(configuration.getTenantId())) { + return configuration.getTenantId(); } return workflowModuleId; @@ -101,7 +101,7 @@ public static class AdapterConfiguration extends WorkerProperties { private boolean useTenants = true; - private String tenant; + private String tenantId; public boolean isUseTenants() { return useTenants; @@ -111,12 +111,12 @@ public void setUseTenants(boolean useTenants) { this.useTenants = useTenants; } - public String getTenant() { - return tenant; + public String getTenantId() { + return tenantId; } - public void setTenant(String tenant) { - this.tenant = tenant; + public void setTenantId(String tenantId) { + this.tenantId = tenantId; } } diff --git a/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8ProcessService.java b/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8ProcessService.java index f058308..0c7eda7 100644 --- a/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8ProcessService.java +++ b/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8ProcessService.java @@ -5,15 +5,18 @@ import io.vanillabp.camunda8.Camunda8VanillaBpProperties; import io.vanillabp.springboot.adapter.AdapterAwareProcessService; import io.vanillabp.springboot.adapter.ProcessServiceImplementation; +import java.time.Duration; +import java.util.Collection; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.function.Function; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.data.repository.CrudRepository; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; - -import java.util.Collection; -import java.util.concurrent.TimeUnit; -import java.util.function.Function; +import org.springframework.transaction.support.TransactionSynchronizationManager; @Transactional(propagation = Propagation.MANDATORY) public class Camunda8ProcessService @@ -29,18 +32,22 @@ public class Camunda8ProcessService private final Camunda8VanillaBpProperties camunda8Properties; + private final ApplicationEventPublisher publisher; + private AdapterAwareProcessService parent; - + private ZeebeClient client; - + public Camunda8ProcessService( final Camunda8VanillaBpProperties camunda8Properties, + final ApplicationEventPublisher publisher, final CrudRepository workflowAggregateRepository, final Function getWorkflowAggregateId, final Class workflowAggregateClass) { super(); this.camunda8Properties = camunda8Properties; + this.publisher = publisher; this.workflowAggregateRepository = workflowAggregateRepository; this.workflowAggregateClass = workflowAggregateClass; this.getWorkflowAggregateId = getWorkflowAggregateId; @@ -98,31 +105,34 @@ public CrudRepository getWorkflowAggregateRepository() { @Override public DE startWorkflow( final DE workflowAggregate) throws Exception { - - // persist to get ID in case of @Id @GeneratedValue - // or force optimistic locking exceptions before running - // the workflow if aggregate was already persisted before - final var attachedAggregate = workflowAggregateRepository - .save(workflowAggregate); - - final var tenantId = camunda8Properties.getTenantId(parent.getWorkflowModuleId()); - final var command = client - .newCreateInstanceCommand() - .bpmnProcessId(parent.getPrimaryBpmnProcessId()) - .latestVersion() - .variables(attachedAggregate); - (tenantId == null - ? command - : command.tenantId(tenantId)) - .send() - .get(10, TimeUnit.SECONDS); + return runInTransaction( + workflowAggregate, + attachedAggregate -> { + final var tenantId = camunda8Properties.getTenantId(parent.getWorkflowModuleId()); + final var command = client + .newCreateInstanceCommand() + .bpmnProcessId(parent.getPrimaryBpmnProcessId()) + .latestVersion() + .variables(attachedAggregate); - try { - return attachedAggregate; - } catch (RuntimeException exception) { - throw exception; - } + try { + (tenantId == null + ? command + : command.tenantId(tenantId)) + .send() + .get(10, TimeUnit.SECONDS); + } catch (Exception e) { + throw new RuntimeException( + "Starting workflow '" + + parent.getPrimaryBpmnProcessId() + + "‘ for aggregate '" + + attachedAggregate + + "' failed!", + e); + } + }, + "startWorkflow"); } @@ -130,19 +140,20 @@ public DE startWorkflow( public DE correlateMessage( final DE workflowAggregate, final String messageName) { - - final var attachedAggregate = workflowAggregateRepository - .save(workflowAggregate); - final var correlationId = getWorkflowAggregateId - .apply(workflowAggregate); - - correlateMessage( + + return runInTransaction( workflowAggregate, - messageName, - correlationId.toString()); - - return attachedAggregate; - + attachedAggregate -> { + final var correlationId = getWorkflowAggregateId + .apply(workflowAggregate); + + doCorrelateMessage( + workflowAggregate, + messageName, + correlationId.toString()); + }, + "correlateMessage"); + } @Override @@ -161,12 +172,21 @@ public DE correlateMessage( final DE workflowAggregate, final String messageName, final String correlationId) { - - // persist to get ID in case of @Id @GeneratedValue - // and force optimistic locking exceptions before running - // the workflow if aggregate was already persisted before - final var attachedAggregate = workflowAggregateRepository - .save(workflowAggregate); + + return runInTransaction( + workflowAggregate, + attachedAggregate -> doCorrelateMessage( + attachedAggregate, + messageName, + correlationId), + "correlateMessage-by-correlationId"); + + } + + private void doCorrelateMessage( + final DE attachedAggregate, + final String messageName, + final String correlationId) { final var tenantId = camunda8Properties.getTenantId(parent.getWorkflowModuleId()); final var command = client @@ -188,8 +208,6 @@ public DE correlateMessage( parent.getPrimaryBpmnProcessId(), messageKey); - return attachedAggregate; - } @Override @@ -209,23 +227,23 @@ public DE correlateMessage( public DE completeTask( final DE workflowAggregate, final String taskId) { - - // force optimistic locking exceptions before running the workflow - final var attachedAggregate = workflowAggregateRepository - .save(workflowAggregate); - - client - .newCompleteCommand(Long.parseLong(taskId, 16)) - .variables(attachedAggregate) - .send() - .join(); - logger.trace("Complete usertask '{}' for process '{}'", + return runInTransaction( + workflowAggregate, taskId, - parent.getPrimaryBpmnProcessId()); - - return attachedAggregate; - + attachedAggregate -> { + client + .newCompleteCommand(Long.parseLong(taskId, 16)) + .variables(attachedAggregate) + .send() + .join(); + + logger.trace("Complete task '{}' of process '{}'", + taskId, + parent.getPrimaryBpmnProcessId()); + }, + "completeTask"); + } @Override @@ -233,7 +251,21 @@ public DE completeUserTask( final DE workflowAggregate, final String taskId) { - return completeTask(workflowAggregate, taskId); + return runInTransaction( + workflowAggregate, + taskId, + attachedAggregate -> { + client + .newCompleteCommand(Long.parseLong(taskId, 16)) + .variables(attachedAggregate) + .send() + .join(); + + logger.trace("Complete user task '{}' of process '{}'", + taskId, + parent.getPrimaryBpmnProcessId()); + }, + "completeUserTask"); } @@ -243,22 +275,22 @@ public DE cancelTask( final String taskId, final String errorCode) { - // force optimistic locking exceptions before running the workflow - final var attachedAggregate = workflowAggregateRepository - .save(workflowAggregate); - - client - .newThrowErrorCommand(Long.parseLong(taskId)) - .errorCode(errorCode) - .send() - .join(); - - logger.trace("Complete usertask '{}' for process '{}'", + return runInTransaction( + workflowAggregate, taskId, - parent.getPrimaryBpmnProcessId()); - - return attachedAggregate; - + attachedAggregate -> { + client + .newThrowErrorCommand(Long.parseLong(taskId)) + .errorCode(errorCode) + .send() + .join(); + + logger.trace("Complete task '{}' of process '{}'", + taskId, + parent.getPrimaryBpmnProcessId()); + }, + "cancelTask"); + } @Override @@ -271,4 +303,54 @@ public DE cancelUserTask( } + private DE runInTransaction( + final DE workflowAggregate, + final Consumer runnable, + final String methodSignature) { + + return runInTransaction( + workflowAggregate, + null, + runnable, + methodSignature); + + } + + private DE runInTransaction( + final DE workflowAggregate, + final String taskIdToTestForAlreadyCompletedOrCancelled, + final Consumer runnable, + final String methodSignature) { + + // persist to get ID in case of @Id @GeneratedValue + // or force optimistic locking exceptions before running + // the workflow if aggregate was already persisted before + final var attachedAggregate = workflowAggregateRepository + .save(workflowAggregate); + + if (TransactionSynchronizationManager.isActualTransactionActive()) { + if (taskIdToTestForAlreadyCompletedOrCancelled != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8TestForTaskAlreadyCompletedOrCancelled( + methodSignature, + () -> client + .newUpdateTimeoutCommand(Long.parseUnsignedLong(taskIdToTestForAlreadyCompletedOrCancelled, 16)) + .timeout(Duration.ofMinutes(10)) + .send() + .join(5, TimeUnit.MINUTES), // needs to run synchronously + () -> "aggregate: " + getWorkflowAggregateId.apply(attachedAggregate) + "; bpmn-process-id: " + parent.getPrimaryBpmnProcessId())); + } + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8CommandAfterTx( + methodSignature, + () -> runnable.accept(attachedAggregate), + () -> "aggregate: " + getWorkflowAggregateId.apply(attachedAggregate) + "; bpmn-process-id: " + parent.getPrimaryBpmnProcessId())); + } else { + runnable.accept(attachedAggregate); + } + + return attachedAggregate; + + } + } diff --git a/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8TransactionAspect.java b/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8TransactionAspect.java new file mode 100644 index 0000000..b5ffa45 --- /dev/null +++ b/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8TransactionAspect.java @@ -0,0 +1,241 @@ +package io.vanillabp.camunda8.service; + +import io.vanillabp.spi.service.TaskException; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; +import org.aspectj.lang.ProceedingJoinPoint; +import org.aspectj.lang.annotation.Around; +import org.aspectj.lang.annotation.Aspect; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +@Aspect +public class Camunda8TransactionAspect { + + private static final Logger logger = LoggerFactory.getLogger(Camunda8TransactionAspect.class); + + public static class TaskHandlerActions { + public Supplier>> testForTaskAlreadyCompletedOrCancelledCommand; + public Map.Entry, Function> bpmnErrorCommand; + public Map.Entry, Function> handlerFailedCommand; + public Supplier>> handlerCompletedCommand; + } + + public static class RunDeferredInTransaction { + public RunDeferredInTransactionSupplier[] argsSupplier; + public Runnable saveAggregateAfterWorkflowTask; + } + + public interface RunDeferredInTransactionSupplier extends Supplier { } + + public static final ThreadLocal actions = ThreadLocal.withInitial(TaskHandlerActions::new); + + public static final ThreadLocal runDeferredInTransaction = ThreadLocal.withInitial(RunDeferredInTransaction::new); + + private final ApplicationEventPublisher publisher; + + public Camunda8TransactionAspect( + final ApplicationEventPublisher publisher) { + + this.publisher = publisher; + + } + + public static void registerDeferredInTransaction( + final RunDeferredInTransactionSupplier[] argsSupplier, + final Runnable saveAggregateAfterWorkflowTask) { + + runDeferredInTransaction.get().argsSupplier = argsSupplier; + runDeferredInTransaction.get().saveAggregateAfterWorkflowTask = saveAggregateAfterWorkflowTask; + + } + + public static void unregisterDeferredInTransaction() { + + runDeferredInTransaction.get().argsSupplier = null; + runDeferredInTransaction.get().saveAggregateAfterWorkflowTask = null; + + } + + private void saveWorkflowAggregate() { + + runDeferredInTransaction.get().saveAggregateAfterWorkflowTask.run(); + + } + + @Around("@annotation(io.vanillabp.spi.service.WorkflowTask)") + private Object checkForTransaction( + final ProceedingJoinPoint pjp) throws Throwable { + + final var methodSignature = pjp.getSignature().toLongString(); + + final var isTxActive = TransactionSynchronizationManager.isActualTransactionActive(); + + try { + + final var newArgs = runDeferredInTransactionArgsSupplier(pjp.getArgs()); + final var value = pjp.proceed(newArgs); // run @WorkflowTask annotated method + saveWorkflowAggregate(); + + if (isTxActive + && (actions.get().testForTaskAlreadyCompletedOrCancelledCommand != null)) { + final var handlerTestCommand = actions.get().testForTaskAlreadyCompletedOrCancelledCommand.get(); + if (handlerTestCommand != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8TestForTaskAlreadyCompletedOrCancelled( + methodSignature, + handlerTestCommand.getKey(), + handlerTestCommand.getValue())); + } + } + if (actions.get().handlerCompletedCommand != null) { + final var handlerCompletedCommand = actions.get().handlerCompletedCommand.get(); + if (handlerCompletedCommand != null) { + if (isTxActive) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8CommandAfterTx( + methodSignature, + handlerCompletedCommand.getKey(), + handlerCompletedCommand.getValue())); + } else { + try { + handlerCompletedCommand.getKey().run(); + } catch (Exception e) { + final var description = handlerCompletedCommand.getValue(); + if (description != null) { + logger.error( + "Could not execute '{}'! Manual action required!", + description.get(), + e); + } else { + logger.error( + "Manual action required due to:", + e); + } + } + } + } + } + return value; + + } catch (TaskException taskError) { + + if (isTxActive + && (actions.get().testForTaskAlreadyCompletedOrCancelledCommand != null)) { + final var handlerTestCommand = actions.get().testForTaskAlreadyCompletedOrCancelledCommand.get(); + if (handlerTestCommand != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8TestForTaskAlreadyCompletedOrCancelled( + methodSignature, + handlerTestCommand.getKey(), + handlerTestCommand.getValue())); + } + } + if (actions.get().bpmnErrorCommand != null) { + final var runnable = actions.get().bpmnErrorCommand.getKey(); + final var description = actions.get().bpmnErrorCommand.getValue(); + if (isTxActive) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8CommandAfterTx( + methodSignature, + () -> runnable.accept(taskError), + () -> description.apply(taskError))); + } else { + try { + runnable.accept(taskError); + } catch (Exception e) { + if (description != null) { + logger.error( + "Could not execute '{}'! Manual action required!", + description.apply(taskError), + e); + } else { + logger.error( + "Manual action required due to:", + e); + } + } + } + } + return null; + + } catch (Exception e) { + + if (isTxActive + && (actions.get().testForTaskAlreadyCompletedOrCancelledCommand != null)) { + final var handlerTestCommand = actions.get().testForTaskAlreadyCompletedOrCancelledCommand.get(); + if (handlerTestCommand != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8TestForTaskAlreadyCompletedOrCancelled( + methodSignature, + handlerTestCommand.getKey(), + handlerTestCommand.getValue())); + } + } + if (actions.get().handlerFailedCommand != null) { + final var runnable = actions.get().handlerFailedCommand.getKey(); + final var description = actions.get().handlerFailedCommand.getValue(); + if (isTxActive) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8CommandAfterTx( + methodSignature, + () -> runnable.accept(e), + () -> description.apply(e))); + } else { + try { + runnable.accept(e); + } catch (Exception ie) { + if (description != null) { + logger.error( + "Could not execute '{}'! Manual action required!", + description.apply(e), + ie); + } else { + logger.error( + "Manual action required due to:", + ie); + } + } + } + } + throw e; + + } + + } + + public static void clearCallbacks() { + + actions.get().bpmnErrorCommand = null; + actions.get().handlerCompletedCommand = null; + actions.get().handlerFailedCommand = null; + actions.get().testForTaskAlreadyCompletedOrCancelledCommand = null; + + } + + private Object[] runDeferredInTransactionArgsSupplier( + final Object[] originalArgs) { + + if (originalArgs == null) { + return null; + } + + final var newArgs = new Object[ originalArgs.length ]; + for (var i = 0; i < originalArgs.length; ++i) { + final var supplier = runDeferredInTransaction.get().argsSupplier[i]; + if (supplier != null) { + newArgs[i] = supplier.get(); + } else { + newArgs[i] = originalArgs[i]; + } + } + + return newArgs; + + } + +} diff --git a/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8TransactionProcessor.java b/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8TransactionProcessor.java new file mode 100644 index 0000000..56b2321 --- /dev/null +++ b/spring-boot/src/main/java/io/vanillabp/camunda8/service/Camunda8TransactionProcessor.java @@ -0,0 +1,143 @@ +package io.vanillabp.camunda8.service; + +import io.camunda.zeebe.client.api.command.ClientStatusException; +import io.grpc.Status; +import io.vanillabp.spi.service.TaskException; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.context.ApplicationEvent; +import org.springframework.transaction.event.TransactionPhase; +import org.springframework.transaction.event.TransactionalEventListener; + +public class Camunda8TransactionProcessor { + + private static final Logger logger = LoggerFactory.getLogger(Camunda8TransactionProcessor.class); + + public static void registerCallbacks( + final Supplier>> testForTaskAlreadyCompletedOrCancelledCommand, + final Map.Entry, Function> bpmnErrorCommand, + final Map.Entry, Function> handlerFailedCommand, + final Supplier>> handlerCompletedCommand) { + + final var actions = Camunda8TransactionAspect.actions.get(); + actions.testForTaskAlreadyCompletedOrCancelledCommand = testForTaskAlreadyCompletedOrCancelledCommand; + actions.bpmnErrorCommand = bpmnErrorCommand; + actions.handlerFailedCommand = handlerFailedCommand; + actions.handlerCompletedCommand = handlerCompletedCommand; + + } + + public static Map.Entry, Function> bpmnErrorCommandCallback() { + + return Camunda8TransactionAspect + .actions + .get() + .bpmnErrorCommand; + + } + + public static Map.Entry, Function> handlerFailedCommandCallback() { + + return Camunda8TransactionAspect + .actions + .get() + .handlerFailedCommand; + + } + + public static Map.Entry> handlerCompletedCommandCallback() { + + return Camunda8TransactionAspect + .actions + .get() + .handlerCompletedCommand + .get(); + + } + + public static void unregisterCallbacks() { + + Camunda8TransactionAspect.clearCallbacks(); + + } + + public static class Camunda8CommandAfterTx extends ApplicationEvent { + final Supplier description; + final Runnable runnable; + public Camunda8CommandAfterTx( + final Object source, + final Runnable runnable, + final Supplier description) { + super(source); + this.runnable = runnable; + this.description = description; + } + } + + public static class Camunda8TestForTaskAlreadyCompletedOrCancelled extends ApplicationEvent { + final Supplier description; + final Runnable runnable; + public Camunda8TestForTaskAlreadyCompletedOrCancelled( + final Object source, + final Runnable runnable, + final Supplier description) { + super(source); + this.runnable = runnable; + this.description = description; + } + } + + @TransactionalEventListener( + phase = TransactionPhase.BEFORE_COMMIT, + fallbackExecution = true) + public void processPreCommit( + final Camunda8TestForTaskAlreadyCompletedOrCancelled event) { + + try { + logger.trace("Will test for existence of task '{}' initiated by: {}", + event.description.get(), + event.getSource()); + // this runnable will test whether the task still exists + event.runnable.run(); + } catch (Exception e) { + // if the task is completed or cancelled, then the tx is rolled back + if ((e instanceof ClientStatusException clientStatusException) + && (clientStatusException.getStatus().getCode() == Status.NOT_FOUND.getCode())) { + throw new RuntimeException( + "Will rollback because job was already completed/cancelled! Test-command giving status 'NOT_FOUND':\n" + + event.description.get()); + } else { + throw new RuntimeException( + "Will rollback because testing for job '{}' failed! Test-command:\n" + + event.description.get()); + } + } + + } + + @TransactionalEventListener( + phase = TransactionPhase.AFTER_COMMIT, + fallbackExecution = true) + public void processPostCommit( + final Camunda8CommandAfterTx event) { + + try { + logger.trace("Will execute Camunda command for '{}' initiated by: {}", + event.description.get(), + event.getSource()); + // this runnable will instruct Zeebe + event.runnable.run(); + } catch (Exception e) { + logger.error( + "Could not execute camunda command for '{}'! Manual action required!", + event.description.get(), + e); + } + + } + +} diff --git a/spring-boot/src/main/java/io/vanillabp/camunda8/service/NotUsedCamunda8TransactionInterceptor.java b/spring-boot/src/main/java/io/vanillabp/camunda8/service/NotUsedCamunda8TransactionInterceptor.java new file mode 100644 index 0000000..e5be06a --- /dev/null +++ b/spring-boot/src/main/java/io/vanillabp/camunda8/service/NotUsedCamunda8TransactionInterceptor.java @@ -0,0 +1,110 @@ +package io.vanillabp.camunda8.service; + +import io.vanillabp.spi.service.TaskException; +import java.lang.reflect.Method; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.transaction.TransactionManager; +import org.springframework.transaction.interceptor.TransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +public class NotUsedCamunda8TransactionInterceptor extends TransactionInterceptor { + + private final ApplicationEventPublisher publisher; + + public static final ThreadLocal actions = ThreadLocal.withInitial(TaskHandlerActions::new); + + public NotUsedCamunda8TransactionInterceptor( + final TransactionManager ptm, + final TransactionAttributeSource tas, + final ApplicationEventPublisher publisher) { + super(ptm, tas); + this.publisher = publisher; + } + + public static class TaskHandlerActions { + public Map.Entry> testForTaskAlreadyCompletedOrCancelledCommand; + public Map.Entry, Function> bpmnErrorCommand; + public Map.Entry, Function> handlerFailedCommand; + public Supplier>> handlerCompletedCommand; + } + + @Override + protected Object invokeWithinTransaction( + final Method method, + final Class targetClass, + final InvocationCallback invocation) throws Throwable { + + return super.invokeWithinTransaction(method, targetClass, () -> { + if (!TransactionSynchronizationManager.isActualTransactionActive()) { + return invocation.proceedWithInvocation(); + } + try { + logger.info("Before TX"); + final var result = invocation.proceedWithInvocation(); + logger.info("After TX"); + if (actions.get().testForTaskAlreadyCompletedOrCancelledCommand != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8TestForTaskAlreadyCompletedOrCancelled( + NotUsedCamunda8TransactionInterceptor.class, + actions.get().testForTaskAlreadyCompletedOrCancelledCommand.getKey(), + actions.get().testForTaskAlreadyCompletedOrCancelledCommand.getValue())); + } + if (actions.get().handlerCompletedCommand != null) { + final var handlerCompletedCommand = actions.get().handlerCompletedCommand.get(); + if (handlerCompletedCommand != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8CommandAfterTx( + NotUsedCamunda8TransactionInterceptor.class, + handlerCompletedCommand.getKey(), + handlerCompletedCommand.getValue())); + } + } + return result; + } catch (TaskException taskError) { + if (actions.get().testForTaskAlreadyCompletedOrCancelledCommand != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8TestForTaskAlreadyCompletedOrCancelled( + NotUsedCamunda8TransactionInterceptor.class, + actions.get().testForTaskAlreadyCompletedOrCancelledCommand.getKey(), + actions.get().testForTaskAlreadyCompletedOrCancelledCommand.getValue())); + } + if (actions.get().bpmnErrorCommand != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8CommandAfterTx( + NotUsedCamunda8TransactionInterceptor.class, + () -> actions.get().bpmnErrorCommand.getKey().accept(taskError), + () -> actions.get().bpmnErrorCommand.getValue().apply(taskError))); + } + return null; + } catch (Exception e) { + if (actions.get().testForTaskAlreadyCompletedOrCancelledCommand != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8TestForTaskAlreadyCompletedOrCancelled( + NotUsedCamunda8TransactionInterceptor.class, + actions.get().testForTaskAlreadyCompletedOrCancelledCommand.getKey(), + actions.get().testForTaskAlreadyCompletedOrCancelledCommand.getValue())); + } + if (actions.get().handlerFailedCommand != null) { + publisher.publishEvent( + new Camunda8TransactionProcessor.Camunda8CommandAfterTx( + NotUsedCamunda8TransactionInterceptor.class, + () -> actions.get().handlerFailedCommand.getKey().accept(e), + () -> actions.get().handlerFailedCommand.getValue().apply(e))); + } + throw e; + } finally { + actions.get().bpmnErrorCommand = null; + actions.get().handlerCompletedCommand = null; + actions.get().handlerFailedCommand = null; + actions.get().testForTaskAlreadyCompletedOrCancelledCommand = null; + } + }); + + } + +} diff --git a/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8TaskHandler.java b/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8TaskHandler.java index 3d5d39f..57f7b1c 100644 --- a/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8TaskHandler.java +++ b/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8TaskHandler.java @@ -1,43 +1,49 @@ package io.vanillabp.camunda8.wiring; -import io.camunda.zeebe.client.api.command.FinalCommandStep; +import io.camunda.zeebe.client.ZeebeClient; import io.camunda.zeebe.client.api.response.ActivatedJob; import io.camunda.zeebe.client.api.worker.JobClient; import io.camunda.zeebe.client.api.worker.JobHandler; -import io.camunda.zeebe.spring.client.jobhandling.CommandWrapper; -import io.camunda.zeebe.spring.client.jobhandling.DefaultCommandExceptionHandlingStrategy; +import io.vanillabp.camunda8.service.Camunda8TransactionAspect; +import io.vanillabp.camunda8.service.Camunda8TransactionProcessor; import io.vanillabp.camunda8.wiring.Camunda8Connectable.Type; import io.vanillabp.camunda8.wiring.parameters.Camunda8MultiInstanceIndexMethodParameter; import io.vanillabp.camunda8.wiring.parameters.Camunda8MultiInstanceTotalMethodParameter; -import io.vanillabp.spi.service.TaskEvent.Event; +import io.vanillabp.spi.service.MultiInstanceElementResolver; +import io.vanillabp.spi.service.TaskEvent; import io.vanillabp.spi.service.TaskException; import io.vanillabp.springboot.adapter.MultiInstance; import io.vanillabp.springboot.adapter.TaskHandlerBase; import io.vanillabp.springboot.adapter.wiring.WorkflowAggregateCache; import io.vanillabp.springboot.parameters.MethodParameter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.data.repository.CrudRepository; -import org.springframework.transaction.annotation.Transactional; - +import io.vanillabp.springboot.parameters.ResolverBasedMultiInstanceMethodParameter; +import io.vanillabp.springboot.parameters.WorkflowAggregateMethodParameter; import java.lang.reflect.Method; +import java.time.Duration; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; import java.util.function.Function; +import java.util.function.Supplier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.data.repository.CrudRepository; -public class Camunda8TaskHandler extends TaskHandlerBase implements JobHandler { +public class Camunda8TaskHandler extends TaskHandlerBase implements JobHandler, Consumer { private static final Logger logger = LoggerFactory.getLogger(Camunda8TaskHandler.class); - private final DefaultCommandExceptionHandlingStrategy commandExceptionHandlingStrategy; - private final Type taskType; private final String idPropertyName; + private ZeebeClient zeebeClient; + public Camunda8TaskHandler( final Type taskType, - final DefaultCommandExceptionHandlingStrategy commandExceptionHandlingStrategy, final CrudRepository workflowAggregateRepository, final Object bean, final Method method, @@ -46,46 +52,79 @@ public Camunda8TaskHandler( super(workflowAggregateRepository, bean, method, parameters); this.taskType = taskType; - this.commandExceptionHandlingStrategy = commandExceptionHandlingStrategy; this.idPropertyName = idPropertyName; } - + + @Override + public void accept( + final ZeebeClient zeebeClient) { + + this.zeebeClient = zeebeClient; + + } + @Override protected Logger getLogger() { - + return logger; - + } @SuppressWarnings("unchecked") @Override - @Transactional public void handle( final JobClient client, final ActivatedJob job) throws Exception { - CommandWrapper command = null; try { final var businessKey = getVariable(job, idPropertyName); - - logger.trace("Will handle task '{}' of workflow '{}' ('{}') as job '{}'", + + logger.trace("Will handle task '{}' (task-definition '{}‘) of workflow '{}' (instance-id '{}') as job '{}'", job.getElementId(), + job.getType(), + job.getBpmnProcessId(), job.getProcessInstanceKey(), - job.getProcessDefinitionKey(), job.getKey()); - + final var taskIdRetrieved = new AtomicBoolean(false); - + final var workflowAggregateCache = new WorkflowAggregateCache(); + + Camunda8TransactionAspect.registerDeferredInTransaction( + new Camunda8TransactionAspect.RunDeferredInTransactionSupplier[parameters.size()], + saveAggregateAfterWorkflowTask(workflowAggregateCache)); + + // Any callback used in this method is executed in case of no active transaction. + // In case of an active transaction the callbacks are used by the Camunda8TransactionInterceptor. + Camunda8TransactionProcessor.registerCallbacks( + () -> { + if (taskType == Type.USERTASK) { + return null; + } + if (taskIdRetrieved.get()) { // async processing of service-task + return null; + } + return testForTaskWasCompletedOrCancelled(job); + }, + doThrowError(client, job, workflowAggregateCache), + doFailed(client, job), + () -> { + if (taskType == Type.USERTASK) { + return null; + } + if (taskIdRetrieved.get()) { // async processing of service-task + return null; + } + return doComplete(client, job, workflowAggregateCache); + }); + final Function multiInstanceSupplier = multiInstanceVariable -> getVariable(job, multiInstanceVariable); - - final var workflowAggregateCache = new WorkflowAggregateCache(); - + super.execute( workflowAggregateCache, businessKey, - true, + false, // will be done within transaction boundaries (args, param) -> processTaskParameter( args, param, @@ -100,7 +139,7 @@ public void handle( (args, param) -> processTaskEventParameter( args, param, - () -> Event.CREATED), + () -> TaskEvent.Event.CREATED), (args, param) -> processMultiInstanceIndexParameter( args, param, @@ -125,23 +164,13 @@ public void handle( return workflowAggregateCache.workflowAggregate; }, multiInstanceSupplier)); - if ((taskType != Type.USERTASK) - && !taskIdRetrieved.get()) { - command = createCompleteCommand(client, job, workflowAggregateCache.workflowAggregate); - } - } catch (TaskException bpmnError) { - command = createThrowErrorCommand(client, job, bpmnError); - } catch (Exception e) { - logger.error("Failed to execute job '{}'", job.getKey(), e); - command = createFailedCommand(client, job, e); - } - - if (command != null) { - command.executeAsync(); + } finally { + Camunda8TransactionProcessor.unregisterCallbacks(); + Camunda8TransactionAspect.unregisterDeferredInTransaction(); } } - + @Override protected Object getMultiInstanceElement( final String name, @@ -149,100 +178,231 @@ protected Object getMultiInstanceElement( return multiInstanceSupplier .apply(name); - + } - + @Override protected Integer getMultiInstanceIndex( final String name, final Function multiInstanceSupplier) { - + return (Integer) multiInstanceSupplier .apply(name + Camunda8MultiInstanceIndexMethodParameter.SUFFIX) - 1; - + } - + @Override protected Integer getMultiInstanceTotal( final String name, final Function multiInstanceSupplier) { - + return (Integer) multiInstanceSupplier .apply(name + Camunda8MultiInstanceTotalMethodParameter.SUFFIX); - + } - + @Override protected MultiInstance getMultiInstance( final String name, final Function multiInstanceSupplier) { - + return new MultiInstance( getMultiInstanceElement(name, multiInstanceSupplier), getMultiInstanceTotal(name, multiInstanceSupplier), getMultiInstanceIndex(name, multiInstanceSupplier)); - + } - + private Object getVariable( final ActivatedJob job, final String name) { - + return job .getVariablesAsMap() .get(name); - + + } + + public Runnable saveAggregateAfterWorkflowTask( + final WorkflowAggregateCache aggregateCache) { + + return () -> { + if (aggregateCache.workflowAggregate != null) { + workflowAggregateRepository.save(aggregateCache.workflowAggregate); + } + }; + + } + + @SuppressWarnings("unchecked") + public Map.Entry> testForTaskWasCompletedOrCancelled( + final ActivatedJob job) { + + return Map.entry( + () -> zeebeClient + .newUpdateTimeoutCommand(job) + .timeout(Duration.ofMinutes(10)) + .send() + .join(5, TimeUnit.MINUTES) + , // needs to run synchronously + () -> "update timeout (BPMN: " + job.getBpmnProcessId() + + "; Element: " + job.getElementId() + + "; Task-Definition: " + job.getType() + + "; Process-Instance: " + job.getProcessInstanceKey() + + "; Job: " + job.getKey() + + ")"); + } @SuppressWarnings("unchecked") - public CommandWrapper createCompleteCommand( + public Map.Entry> doComplete( final JobClient jobClient, final ActivatedJob job, - final Object workflowAggregateId) { + final WorkflowAggregateCache workflowAggregateCache) { - var completeCommand = jobClient - .newCompleteCommand(job.getKey()); - - if (workflowAggregateId != null) { - completeCommand = completeCommand.variables(workflowAggregateId); - } - - return new CommandWrapper( - (FinalCommandStep) ((FinalCommandStep) completeCommand), - job, - commandExceptionHandlingStrategy); + return Map.entry( + () -> { + var completeCommand = jobClient + .newCompleteCommand(job.getKey()); + + if (workflowAggregateCache.workflowAggregate != null) { + completeCommand = completeCommand.variables(workflowAggregateCache.workflowAggregate); + } + + completeCommand + .send() + .exceptionally(t -> { + throw new RuntimeException("error", t); + }); + }, + () -> "complete command (BPMN: " + job.getBpmnProcessId() + + "; Element: " + job.getElementId() + + "; Task-Definition: " + job.getType() + + "; Process-Instance: " + job.getProcessInstanceKey() + + "; Job: " + job.getKey() + + ")"); } - private CommandWrapper createThrowErrorCommand( + private Map.Entry, Function> doThrowError( final JobClient jobClient, final ActivatedJob job, - final TaskException bpmnError) { + final WorkflowAggregateCache workflowAggregateCache) { - return new CommandWrapper( - jobClient - .newThrowErrorCommand(job.getKey()) - .errorCode(bpmnError.getErrorCode()) - .errorMessage(bpmnError.getErrorName()), - job, - commandExceptionHandlingStrategy); + return Map.entry( + taskException -> { + var throwErrorCommand = jobClient + .newThrowErrorCommand(job.getKey()) + .errorCode(taskException.getErrorCode()) + .errorMessage(taskException.getErrorName()); + if (workflowAggregateCache.workflowAggregate != null) { + throwErrorCommand = throwErrorCommand.variables(workflowAggregateCache.workflowAggregate); + } + + throwErrorCommand + .send() + .exceptionally(t -> { + throw new RuntimeException("error", t); + }); + }, + taskException -> "throw error command (BPMN: " + job.getBpmnProcessId() + + "; Element: " + job.getElementId() + + "; Task-Definition: " + job.getType() + + "; Process-Instance: " + job.getProcessInstanceKey() + + "; Job: " + job.getKey() + + ")"); } - + @SuppressWarnings("unchecked") - private CommandWrapper createFailedCommand( + private Map.Entry, Function> doFailed( final JobClient jobClient, - final ActivatedJob job, - final Exception e) { - - return new CommandWrapper( - (FinalCommandStep) ((FinalCommandStep) jobClient - .newFailCommand(job) - .retries(0) - .errorMessage(e.getMessage())), - job, - commandExceptionHandlingStrategy); - + final ActivatedJob job) { + + return Map.entry( + exception -> { + jobClient + .newFailCommand(job) + .retries(0) + .errorMessage(exception.getMessage()) + .send() + .exceptionally(t -> { + throw new RuntimeException("error", t); + }); + }, + taskException -> "fail command (BPMN: " + job.getBpmnProcessId() + + "; Element: " + job.getElementId() + + "; Task-Definition: " + job.getType() + + "; Process-Instance: " + job.getProcessInstanceKey() + + "; Job: " + job.getKey() + + ")"); + + } + + protected boolean processWorkflowAggregateParameter( + final Object[] args, + final MethodParameter param, + final WorkflowAggregateCache workflowAggregateCache, + final Object workflowAggregateId) { + + if (!(param instanceof WorkflowAggregateMethodParameter)) { + return true; + } + + Camunda8TransactionAspect.runDeferredInTransaction.get().argsSupplier[param.getIndex()] = () -> { + // Using findById is required to get an object instead of a Hibernate proxy. + // Otherwise for e.g. Camunda8 connector JSON serialization of the + // workflow aggregate is not possible. + workflowAggregateCache.workflowAggregate = workflowAggregateRepository + .findById(workflowAggregateId) + .orElse(null); + return workflowAggregateCache.workflowAggregate; + }; + + args[param.getIndex()] = null; // will be set by deferred execution of supplier + + return false; + + } + + protected boolean processMultiInstanceResolverParameter( + final Object[] args, + final MethodParameter param, + final Supplier workflowAggregate, + final Function multiInstanceSupplier) { + + if (!(param instanceof ResolverBasedMultiInstanceMethodParameter)) { + return true; + } + + @SuppressWarnings("unchecked") + final var resolver = + (MultiInstanceElementResolver) + ((ResolverBasedMultiInstanceMethodParameter) param).getResolverBean(); + + final var multiInstances = new HashMap>(); + + resolver + .getNames() + .forEach(name -> multiInstances.put(name, getMultiInstance(name, multiInstanceSupplier))); + + Camunda8TransactionAspect.runDeferredInTransaction.get().argsSupplier[param.getIndex()] = () -> { + try { + return resolver.resolve(workflowAggregate.get(), multiInstances); + } catch (Exception e) { + throw new RuntimeException( + "Failed processing MultiInstanceElementResolver for parameter '" + + param.getParameter() + + "' of method '" + + method + + "'", e); + } + }; + + args[param.getIndex()] = null; // will be set by deferred execution of supplier + + return false; + } } diff --git a/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8TaskWiring.java b/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8TaskWiring.java index c3da080..c346cee 100644 --- a/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8TaskWiring.java +++ b/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8TaskWiring.java @@ -55,6 +55,8 @@ public class Camunda8TaskWiring extends TaskWiringBase workers = new LinkedList<>(); + private List handlers = new LinkedList<>(); + private Set userTaskTenantIds = new HashSet<>(); private final Camunda8VanillaBpProperties camunda8Properties; @@ -96,6 +98,7 @@ public void accept( final ZeebeClient client) { this.client = client; + handlers.forEach(handler -> handler.accept(client)); } @@ -239,6 +242,11 @@ protected void connectToBpms( method, parameters, idPropertyName); + if (this.client != null) { + taskHandler.accept(this.client); + } else { + handlers.add(taskHandler); + } if (connectable.getType() == Type.USERTASK) { @@ -251,7 +259,7 @@ protected void connectToBpms( return; } - + final var variablesToFetch = getVariablesToFetch(idPropertyName, parameters); final var worker = client .newWorker() diff --git a/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8UserTaskHandler.java b/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8UserTaskHandler.java index 7daa7ec..c75b2a9 100644 --- a/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8UserTaskHandler.java +++ b/spring-boot/src/main/java/io/vanillabp/camunda8/wiring/Camunda8UserTaskHandler.java @@ -18,7 +18,7 @@ public class Camunda8UserTaskHandler implements JobHandler { private final Map taskHandlers = new HashMap<>(); - private String workerId; + private final String workerId; public Camunda8UserTaskHandler( final String workerId) { diff --git a/spring-boot/src/main/resources/io/vanillabp/camunda8/liquibase/initial_setup.yaml b/spring-boot/src/main/resources/io/vanillabp/camunda8/liquibase/initial_setup.yaml index ef56bbe..c98fd6b 100644 --- a/spring-boot/src/main/resources/io/vanillabp/camunda8/liquibase/initial_setup.yaml +++ b/spring-boot/src/main/resources/io/vanillabp/camunda8/liquibase/initial_setup.yaml @@ -2,7 +2,7 @@ databaseChangeLog: - changeSet: id: initial_setup.yaml author: stephanpelikan - dbms: "!oracle" + dbms: "h2" changes: - createTable: tableName: CAMUNDA8_RESOURCES diff --git a/spring-boot/src/main/resources/io/vanillabp/camunda8/liquibase/issue_26.yaml b/spring-boot/src/main/resources/io/vanillabp/camunda8/liquibase/issue_26.yaml index f846ca2..dd7498f 100644 --- a/spring-boot/src/main/resources/io/vanillabp/camunda8/liquibase/issue_26.yaml +++ b/spring-boot/src/main/resources/io/vanillabp/camunda8/liquibase/issue_26.yaml @@ -2,7 +2,7 @@ databaseChangeLog: - changeSet: id: issue_26.yaml#initial_setup author: stephanpelikan - dbms: oracle + dbms: "!h2" changes: - createTable: tableName: CAMUNDA8_RESOURCES @@ -108,7 +108,7 @@ databaseChangeLog: - changeSet: id: issue_26.yaml#rename_columns author: stephanpelikan - dbms: "!oracle" + dbms: h2 changes: - renameColumn: tableName: CAMUNDA8_RESOURCES @@ -162,3 +162,11 @@ databaseChangeLog: tableName: CAMUNDA8_DEPLOYMENTS oldColumnName: BPMN_PROCESS_ID newColumnName: C8D_BPMN_PROCESS_ID + - changeSet: + id: issue_26.yaml#change_type_of_definition_key + author: stephanpelikan + changes: + - modifyDataType: + columnName: C8D_DEFINITION_KEY + newDataType: bigint + tableName: CAMUNDA8_DEPLOYMENTS