Add Retryable attribute for FunctionBinding annotation

activiti-cloud-api/pom.xml

@@ -12,7 +12,7 @@
   <name>Activiti Cloud :: Runtime API Parent</name>
   <packaging>pom</packaging>
   <properties>
-    <activiti.version>7.10.0-alpha.1</activiti.version>
+    <activiti.version>0.0.1-PR-4264-1227-SNAPSHOT</activiti.version>
   </properties>
   <modules>
     <module>activiti-cloud-api-dependencies</module>

activiti-cloud-modeling-service/pom.xml

@@ -19,7 +19,7 @@
     <module>activiti-cloud-acceptance-tests-modeling</module>
   </modules>
   <properties>
-    <activiti.version>7.10.0-alpha.1</activiti.version>
+    <activiti.version>0.0.1-PR-4264-1227-SNAPSHOT</activiti.version>
     <everit-json-schema.version>1.14.0</everit-json-schema.version>
     <commons-collections4.version>4.4</commons-collections4.version>
     <json-java.version>20230227</json-java.version>

activiti-cloud-query-service/pom.xml

@@ -18,7 +18,7 @@
     <module>activiti-cloud-starter-query</module>
   </modules>
   <properties>
-    <activiti.version>7.10.0-alpha.1</activiti.version>
+    <activiti.version>0.0.1-PR-4264-1227-SNAPSHOT</activiti.version>
   </properties>
   <!-- BoM Marker Dependencies -->
   <dependencies>

activiti-cloud-runtime-bundle-service/activiti-cloud-services-runtime-bundle/activiti-cloud-services-connectors/src/main/java/org/activiti/services/connectors/channel/RuntimeIntegrationEventInputBinding.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2017-2020 Alfresco Software, Ltd.
+ *
+ * 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.activiti.services.connectors.channel;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import org.activiti.cloud.common.messaging.functional.FunctionBinding;
+import org.activiti.engine.ActivitiOptimisticLockingException;
+import org.springframework.core.annotation.AliasFor;
+import org.springframework.retry.annotation.Backoff;
+import org.springframework.retry.annotation.Retryable;
+
+@Retention( RetentionPolicy.RUNTIME )
+@Target( {ElementType.METHOD, ElementType.TYPE} )
+@FunctionBinding(retryable = @Retryable(value = ActivitiOptimisticLockingException.class,
+                                        maxAttemptsExpression = "${activiti.cloud.runtime.integration.retryable.max-attempts:3}",
+                                        backoff = @Backoff(delayExpression = "${activiti.cloud.runtime.integration.retryable.backoff.delay:0}")))
+public @interface RuntimeIntegrationEventInputBinding {
+    @AliasFor(annotation = FunctionBinding.class, attribute = "input")
+    String value();
+}

activiti-cloud-runtime-bundle-service/activiti-cloud-services-runtime-bundle/activiti-cloud-services-connectors/src/main/java/org/activiti/services/connectors/channel/ServiceTaskIntegrationErrorEventHandler.java

@@ -16,22 +16,24 @@
 
 package org.activiti.services.connectors.channel;
 
+import java.util.ArrayList;
+import java.util.List;
 import org.activiti.api.process.model.IntegrationContext;
 import org.activiti.cloud.api.process.model.CloudBpmnError;
 import org.activiti.cloud.api.process.model.IntegrationError;
 import org.activiti.cloud.services.events.configuration.RuntimeBundleProperties;
 import org.activiti.cloud.services.events.listeners.ProcessEngineEventsAggregator;
 import org.activiti.engine.ManagementService;
 import org.activiti.engine.RuntimeService;
+import org.activiti.engine.impl.cmd.integration.DeleteIntegrationContextCmd;
+import org.activiti.engine.impl.interceptor.Command;
 import org.activiti.engine.impl.persistence.entity.ExecutionEntity;
 import org.activiti.engine.impl.persistence.entity.integration.IntegrationContextEntity;
 import org.activiti.engine.integration.IntegrationContextService;
 import org.activiti.engine.runtime.Execution;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.List;
-
 public class ServiceTaskIntegrationErrorEventHandler {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ServiceTaskIntegrationErrorEventHandler.class);
@@ -59,7 +61,9 @@ public void receive(IntegrationError integrationError) {
         IntegrationContextEntity integrationContextEntity = integrationContextService.findById(integrationContext.getId());
 
         if (integrationContextEntity != null) {
-            integrationContextService.deleteIntegrationContext(integrationContextEntity);
+            List<Command<?>> commands = new ArrayList<>();
+
+            commands.add(new DeleteIntegrationContextCmd(integrationContextEntity));
 
             List<Execution> executions = runtimeService.createExecutionQuery().executionId(integrationContextEntity.getExecutionId()).list();
             if (executions.size() > 0) {
@@ -77,7 +81,13 @@ public void receive(IntegrationError integrationError) {
                 if (CloudBpmnError.class.getName().equals(errorClassName)) {
                     if (execution.getActivityId().equals(clientId)) {
                         try {
-                            triggerIntegrationContextError(integrationError, execution);
+                            commands.add(new PropagateCloudBpmnErrorCmd(integrationError, execution));
+                            commands.add(new AggregateIntegrationErrorReceivedClosingEventCmd(
+                                    new AggregateIntegrationErrorReceivedEventCmd(integrationError,
+                                                                                  runtimeBundleProperties,
+                                                                                  processEngineEventsAggregator)));
+
+                            managementService.executeCommand(CompositeCommand.of(commands.toArray(Command[]::new)));
                             return;
                         } catch (Throwable cause) {
                             LOGGER.error("Error propagating CloudBpmnError: {}", cause.getMessage());
@@ -97,17 +107,11 @@ public void receive(IntegrationError integrationError) {
                 LOGGER.warn(message);
             }
 
-            managementService.executeCommand(new AggregateIntegrationErrorReceivedEventCmd(
-                integrationError, runtimeBundleProperties, processEngineEventsAggregator));
-        }
-    }
+            commands.add(new AggregateIntegrationErrorReceivedEventCmd(integrationError,
+                                                                       runtimeBundleProperties,
+                                                                       processEngineEventsAggregator));
 
-    private void triggerIntegrationContextError(IntegrationError integrationError, ExecutionEntity execution) {
-        managementService.executeCommand(
-            CompositeCommand.of(
-                new PropagateCloudBpmnErrorCmd(integrationError, execution),
-                new AggregateIntegrationErrorReceivedClosingEventCmd(new AggregateIntegrationErrorReceivedEventCmd(
-                    integrationError, runtimeBundleProperties, processEngineEventsAggregator))));
+            managementService.executeCommand(CompositeCommand.of(commands.toArray(Command[]::new)));
+        }
     }
-
 }

activiti-cloud-runtime-bundle-service/activiti-cloud-services-runtime-bundle/activiti-cloud-services-connectors/src/main/java/org/activiti/services/connectors/channel/ServiceTaskIntegrationResultEventHandler.java

@@ -16,6 +16,7 @@
 
 package org.activiti.services.connectors.channel;
 
+import java.util.ArrayList;
 import java.util.List;
 import org.activiti.api.process.model.IntegrationContext;
 import org.activiti.cloud.api.process.model.IntegrationResult;
@@ -25,6 +26,8 @@
 import org.activiti.engine.RuntimeService;
 import org.activiti.engine.impl.bpmn.behavior.VariablesPropagator;
 import org.activiti.engine.impl.cmd.TriggerCmd;
+import org.activiti.engine.impl.cmd.integration.DeleteIntegrationContextCmd;
+import org.activiti.engine.impl.interceptor.Command;
 import org.activiti.engine.impl.persistence.entity.integration.IntegrationContextEntity;
 import org.activiti.engine.integration.IntegrationContextService;
 import org.activiti.engine.runtime.Execution;
@@ -60,20 +63,23 @@ public void receive(IntegrationResult integrationResult) {
         IntegrationContext integrationContext = integrationResult.getIntegrationContext();
         IntegrationContextEntity integrationContextEntity = integrationContextService.findById(integrationContext.getId());
 
-        String executionId = integrationContext.getExecutionId();
-        List<Execution> executions = runtimeService.createExecutionQuery()
-                                                   .executionId(executionId)
-                                                   .list();
         if (integrationContextEntity != null) {
-            integrationContextService.deleteIntegrationContext(integrationContextEntity);
+            List<Command<?>> commands = new ArrayList<>();
 
+            commands.add(new DeleteIntegrationContextCmd(integrationContextEntity));
+
+            String executionId = integrationContext.getExecutionId();
+            List<Execution> executions = runtimeService.createExecutionQuery()
+                                                       .executionId(executionId)
+                                                       .list();
             if (executions.size() > 0) {
                 Execution execution = executions.get(0);
 
                 if (execution.getActivityId()
                              .equals(integrationContext.getClientId())) {
-                    triggerIntegrationContextExecution(integrationContext);
-                    return;
+                    commands.add(new TriggerCmd(integrationContext.getExecutionId(),
+                                                integrationContext.getOutBoundVariables(),
+                                                variablesPropagator));
                 } else {
                     LOGGER.warn("Could not find matching activityId '{}' for integration result '{}' with executionId '{}'",
                                 integrationContext.getClientId(),
@@ -87,19 +93,11 @@ public void receive(IntegrationResult integrationResult) {
                     "`. The integration result for the integration context `" + integrationContext.getId() + "` will be ignored.";
                 LOGGER.warn(message);
             }
-            managementService.executeCommand(new AggregateIntegrationResultReceivedEventCmd(
-                integrationContext, runtimeBundleProperties, processEngineEventsAggregator));
-        }
-    }
 
-    private void triggerIntegrationContextExecution(IntegrationContext integrationContext) {
-        managementService.executeCommand(
-            CompositeCommand.of(
-                new TriggerCmd(integrationContext.getExecutionId(), integrationContext.getOutBoundVariables(),
-                    variablesPropagator),
-                new AggregateIntegrationResultReceivedEventCmd(integrationContext,
-                    runtimeBundleProperties, processEngineEventsAggregator)
-            ));
-    }
+            commands.add(new AggregateIntegrationResultReceivedEventCmd(
+                    integrationContext, runtimeBundleProperties, processEngineEventsAggregator));
 
+            managementService.executeCommand(CompositeCommand.of(commands.toArray(Command[]::new)));
+        }
+    }
 }

activiti-cloud-runtime-bundle-service/activiti-cloud-services-runtime-bundle/activiti-cloud-services-connectors/src/main/java/org/activiti/services/connectors/conf/CloudConnectorsAutoConfiguration.java

@@ -19,7 +19,6 @@
 import java.util.function.Consumer;
 import org.activiti.cloud.api.process.model.IntegrationError;
 import org.activiti.cloud.api.process.model.IntegrationResult;
-import org.activiti.cloud.common.messaging.functional.FunctionBinding;
 import org.activiti.cloud.services.events.configuration.RuntimeBundleProperties;
 import org.activiti.cloud.services.events.converter.RuntimeBundleInfoAppender;
 import org.activiti.cloud.services.events.listeners.ProcessEngineEventsAggregator;
@@ -37,6 +36,7 @@
 import org.activiti.services.connectors.channel.IntegrationRequestBuilder;
 import org.activiti.services.connectors.channel.IntegrationRequestReplayer;
 import org.activiti.services.connectors.channel.ProcessEngineIntegrationChannels;
+import org.activiti.services.connectors.channel.RuntimeIntegrationEventInputBinding;
 import org.activiti.services.connectors.channel.ServiceTaskIntegrationErrorEventHandler;
 import org.activiti.services.connectors.channel.ServiceTaskIntegrationResultEventHandler;
 import org.activiti.services.connectors.message.IntegrationContextMessageBuilderFactory;
@@ -71,8 +71,8 @@ public ServiceTaskIntegrationResultEventHandler serviceTaskIntegrationResultEven
             runtimeBundleProperties, managementService, processEngineEventsAggregator, variablesPropagator);
     }
 
-    @FunctionBinding(input = ProcessEngineIntegrationChannels.INTEGRATION_RESULTS_CONSUMER)
     @Bean
+    @RuntimeIntegrationEventInputBinding(ProcessEngineIntegrationChannels.INTEGRATION_RESULTS_CONSUMER)
     public Consumer<Message<IntegrationResult>> serviceTaskIntegrationResultEventConsumer(ServiceTaskIntegrationResultEventHandler handler) {
         return message -> handler.receive(message.getPayload());
     }
@@ -91,8 +91,8 @@ public ServiceTaskIntegrationErrorEventHandler serviceTaskIntegrationErrorEventH
                                                            processEngineEventsAggregator);
     }
 
-    @FunctionBinding(input = ProcessEngineIntegrationChannels.INTEGRATION_ERRORS_CONSUMER)
     @Bean
+    @RuntimeIntegrationEventInputBinding(ProcessEngineIntegrationChannels.INTEGRATION_ERRORS_CONSUMER)
     public Consumer<Message<IntegrationError>> serviceTaskIntegrationErrorEventConsumer(ServiceTaskIntegrationErrorEventHandler handler) {
         return message -> handler.receive(message.getPayload());
     }

activiti-cloud-runtime-bundle-service/activiti-cloud-services-runtime-bundle/activiti-cloud-services-connectors/src/test/java/org/activiti/services/connectors/channel/ServiceTaskIntegrationErrorEventHandlerTest.java

@@ -15,19 +15,15 @@
  */
 package org.activiti.services.connectors.channel;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import java.util.Collections;
 import org.activiti.api.runtime.model.impl.IntegrationContextImpl;
 import org.activiti.cloud.api.process.model.CloudBpmnError;
 import org.activiti.cloud.api.process.model.IntegrationError;
 import org.activiti.cloud.api.process.model.impl.IntegrationErrorImpl;
 import org.activiti.cloud.api.process.model.impl.IntegrationRequestImpl;
 import org.activiti.engine.ManagementService;
 import org.activiti.engine.RuntimeService;
-import org.activiti.engine.impl.interceptor.Command;
+import org.activiti.engine.impl.cmd.integration.DeleteIntegrationContextCmd;
 import org.activiti.engine.impl.persistence.entity.ExecutionEntity;
 import org.activiti.engine.impl.persistence.entity.integration.IntegrationContextEntityImpl;
 import org.activiti.engine.integration.IntegrationContextService;
@@ -41,7 +37,11 @@
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
-import java.util.Collections;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
 public class ServiceTaskIntegrationErrorEventHandlerTest {
@@ -60,7 +60,7 @@ public class ServiceTaskIntegrationErrorEventHandlerTest {
     private IntegrationContextService integrationContextService;
 
     @Captor
-    private ArgumentCaptor<Command<?>> commandArgumentCaptor;
+    private ArgumentCaptor<CompositeCommand> commandArgumentCaptor;
 
     @Mock
     private ExecutionQuery executionQuery;
@@ -96,13 +96,12 @@ public void should_propagateErrorAndAggregateEvent_when_clientIdMatches() {
         handler.receive(integrationErrorEvent);
 
         //then
-        verify(integrationContextService).deleteIntegrationContext(integrationContextEntity);
         verify(managementService).executeCommand(commandArgumentCaptor.capture());
-        final Command<?> command = commandArgumentCaptor.getValue();
-        assertThat(command).isExactlyInstanceOf(CompositeCommand.class);
-        CompositeCommand compositeCommand = (CompositeCommand) command;
-        assertThat(compositeCommand.getCommands().get(0)).isInstanceOf(PropagateCloudBpmnErrorCmd.class);
-        assertThat(compositeCommand.getCommands().get(1)).isInstanceOf(AggregateIntegrationErrorReceivedClosingEventCmd.class);
+        final CompositeCommand compositeCommand = commandArgumentCaptor.getValue();
+        assertThat(compositeCommand.getCommands()).hasSize(3);
+        assertThat(compositeCommand.getCommands().get(0)).isInstanceOf(DeleteIntegrationContextCmd.class);
+        assertThat(compositeCommand.getCommands().get(1)).isInstanceOf(PropagateCloudBpmnErrorCmd.class);
+        assertThat(compositeCommand.getCommands().get(2)).isInstanceOf(AggregateIntegrationErrorReceivedClosingEventCmd.class);
     }
 
     @Test
@@ -126,10 +125,11 @@ public void should_AggregateEventButNotPropagateError_when_clientIdDoesNotMatch(
         handler.receive(integrationErrorEvent);
 
         //then
-        verify(integrationContextService).deleteIntegrationContext(integrationContextEntity);
         verify(managementService).executeCommand(commandArgumentCaptor.capture());
-        final Command<?> command = commandArgumentCaptor.getValue();
-        assertThat(command).isExactlyInstanceOf(AggregateIntegrationErrorReceivedEventCmd.class);
+        final CompositeCommand compositeCommand = commandArgumentCaptor.getValue();
+        assertThat(compositeCommand.getCommands()).hasSize(2);
+        assertThat(compositeCommand.getCommands().get(0)).isInstanceOf(DeleteIntegrationContextCmd.class);
+        assertThat(compositeCommand.getCommands().get(1)).isInstanceOf(AggregateIntegrationErrorReceivedEventCmd.class);
     }
 
     private IntegrationContextEntityImpl buildIntegrationContextEntity() {