Propagate authenticated user when send-event task is sent asynchronously

The async send-event job re-evaluates eventInParameter source expressions
in a worker thread, where the Authentication thread-local is not set. As a
result expressions like ${authenticatedUserId} resolve to null even though
they would resolve correctly when the same task is configured with
flowable:sendSynchronously=true.

Capture the authenticated user when the async job is scheduled (storing it
on the JobEntity's jobHandlerConfiguration, which the handler does not
otherwise use) and restore it in AsyncSendEventJobHandler before invoking
the activity behavior. The previous authenticated user is restored in a
finally block.

Author: kevinklever

modules/flowable-engine/src/main/java/org/flowable/engine/impl/bpmn/behavior/SendEventTaskActivityBehavior.java

@@ -30,6 +30,7 @@
 import org.flowable.common.engine.api.delegate.Expression;
 import org.flowable.common.engine.api.scope.ScopeTypes;
 import org.flowable.common.engine.impl.el.ExpressionManager;
+import org.flowable.common.engine.impl.identity.Authentication;
 import org.flowable.common.engine.impl.interceptor.CommandContext;
 import org.flowable.engine.ProcessEngineConfiguration;
 import org.flowable.engine.delegate.DelegateExecution;
@@ -92,9 +93,16 @@ public void execute(DelegateExecution execution) {
         boolean sendSynchronously = sendEventServiceTask.isSendSynchronously() || executedAsAsyncJob;
         if (!sendSynchronously) {
             JobService jobService = processEngineConfiguration.getJobServiceConfiguration().getJobService();
-            
+
             JobEntity job = JobUtil.createJob(executionEntity, sendEventServiceTask, AsyncSendEventJobHandler.TYPE, processEngineConfiguration);
 
+            // Capture the currently authenticated user so that ${authenticatedUserId} references in
+            // eventInParameters resolve to the scheduling user when the async job runs in a worker thread.
+            String authenticatedUserId = Authentication.getAuthenticatedUserId();
+            if (StringUtils.isNotEmpty(authenticatedUserId)) {
+                job.setJobHandlerConfiguration(authenticatedUserId);
+            }
+
             jobService.createAsyncJob(job, true);
             jobService.scheduleAsyncJob(job);
 

modules/flowable-engine/src/main/java/org/flowable/engine/impl/jobexecutor/AsyncSendEventJobHandler.java

@@ -12,9 +12,11 @@
  */
 package org.flowable.engine.impl.jobexecutor;
 
+import org.apache.commons.lang3.StringUtils;
 import org.flowable.bpmn.model.FlowElement;
 import org.flowable.bpmn.model.SendEventServiceTask;
 import org.flowable.common.engine.api.FlowableException;
+import org.flowable.common.engine.impl.identity.Authentication;
 import org.flowable.common.engine.impl.interceptor.CommandContext;
 import org.flowable.engine.impl.delegate.ActivityBehavior;
 import org.flowable.engine.impl.persistence.entity.ExecutionEntity;
@@ -50,11 +52,23 @@ public void execute(JobEntity job, String configuration, VariableScope variableS
                     "Unexpected activity behavior (" + behavior.getClass() + ") found for " + job + " at " + executionEntity);
         }
 
+        // Restore the authenticated user that scheduled the send so that ${authenticatedUserId}
+        // references in eventInParameters resolve consistently with synchronous sending.
+        boolean restoreAuthentication = false;
+        String previousAuthenticatedUserId = Authentication.getAuthenticatedUserId();
+        if (StringUtils.isNotEmpty(configuration) && previousAuthenticatedUserId == null) {
+            Authentication.setAuthenticatedUserId(configuration);
+            restoreAuthentication = true;
+        }
+
         try {
             commandContext.addAttribute(TYPE, true); // Will be read in the SendEventTaskActivityBehavior
             activityBehavior.execute(executionEntity);
         } finally {
             commandContext.removeAttribute(TYPE);
+            if (restoreAuthentication) {
+                Authentication.setAuthenticatedUserId(previousAuthenticatedUserId);
+            }
         }
     }
 

modules/flowable-engine/src/test/java/org/flowable/engine/test/eventregistry/SendEventTaskTest.java

@@ -21,6 +21,7 @@
 import java.util.Map;
 
 import org.flowable.common.engine.impl.history.HistoryLevel;
+import org.flowable.common.engine.impl.identity.Authentication;
 import org.flowable.common.engine.impl.interceptor.EngineConfigurationConstants;
 import org.flowable.engine.history.HistoricActivityInstance;
 import org.flowable.engine.impl.jobexecutor.AsyncSendEventJobHandler;
@@ -322,6 +323,50 @@ public void testSendEventWithExpressions() throws Exception {
                         + " }");
     }
 
+    @Test
+    @Deployment
+    public void testSendEventWithAuthenticatedUserExpression() throws Exception {
+        Authentication.setAuthenticatedUserId("alice");
+        try {
+            ProcessInstance processInstance = runtimeService.createProcessInstanceBuilder()
+                    .processDefinitionKey("process")
+                    .variable("accountNumber", 123)
+                    .start();
+
+            assertThat(outboundEventChannelAdapter.receivedEvents).isEmpty();
+
+            Task task = taskService.createTaskQuery().processInstanceId(processInstance.getId()).singleResult();
+            assertThat(task).isNotNull();
+
+            taskService.complete(task.getId());
+
+            Job job = managementService.createJobQuery().processInstanceId(processInstance.getId()).singleResult();
+            assertThat(job).isNotNull();
+            assertThat(job.getJobHandlerType()).isEqualTo(AsyncSendEventJobHandler.TYPE);
+            assertThat(job.getElementId()).isEqualTo("sendEventTask");
+
+            assertThat(outboundEventChannelAdapter.receivedEvents).isEmpty();
+
+            // Clear the authenticated user before the job runs to prove the captured value is restored
+            // by the AsyncSendEventJobHandler instead of relying on a thread-local that the job worker
+            // does not inherit.
+            Authentication.setAuthenticatedUserId(null);
+
+            JobTestHelper.waitForJobExecutorToProcessAllJobs(processEngineConfiguration, managementService, 5000, 200);
+
+            assertThat(outboundEventChannelAdapter.receivedEvents).hasSize(1);
+
+            JsonNode jsonNode = processEngineConfiguration.getObjectMapper().readTree(outboundEventChannelAdapter.receivedEvents.get(0));
+            assertThatJson(jsonNode)
+                    .isEqualTo("{"
+                            + "   nameProperty: 'alice',"
+                            + "   numberProperty: 123"
+                            + " }");
+        } finally {
+            Authentication.setAuthenticatedUserId(null);
+        }
+    }
+
     @Test
     @Deployment
     public void testSendEventSkipExpression() throws Exception {

modules/flowable-engine/src/test/resources/org/flowable/engine/test/eventregistry/SendEventTaskTest.testSendEventWithAuthenticatedUserExpression.bpmn20.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<definitions id="definitions"
+  xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL"
+  xmlns:flowable="http://flowable.org/bpmn"
+  targetNamespace="Examples"
+  xmlns:tns="Examples">
+
+  <process id="process">
+
+    <startEvent id="theStart" />
+
+    <sequenceFlow sourceRef="theStart" targetRef="task" />
+
+    <userTask id="task" />
+
+    <sequenceFlow sourceRef="task" targetRef="sendEventTask" />
+
+    <serviceTask id="sendEventTask" flowable:type="send-event">
+        <extensionElements>
+            <flowable:eventType>anotherEvent</flowable:eventType>
+            <flowable:channelKey>out-channel</flowable:channelKey>
+            <flowable:eventInParameter source="${authenticatedUserId}" target="nameProperty" />
+            <flowable:eventInParameter source="${accountNumber}" target="numberProperty" />
+        </extensionElements>
+    </serviceTask>
+
+    <sequenceFlow sourceRef="sendEventTask" targetRef="taskAfter" />
+
+    <userTask id="taskAfter" />
+
+    <sequenceFlow sourceRef="taskAfter" targetRef="theEnd" />
+
+    <endEvent id="theEnd" />
+
+  </process>
+
+</definitions>