Propagate authenticated user when send-event task is sent asynchronously

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>