Move deploy-time start event registration onto the StartEvent ActivityBehavior

Each process-level start event now owns its full deploy lifecycle via a
new ProcessLevelStartEventActivityBehavior interface (deploy + undeploy),
plus typed deploy/undeploy contexts and four built-in behaviors (Message,
Signal, Timer, EventRegistry). BpmnDeploymentHelper iterates start events
and the EventSubscriptionManager / TimerManager classes are deleted. A
new findEventSubscriptionsByTypesAndProcessDefinitionId finder runs a
single SQL with EVENT_TYPE_ IN (...), so cleanup is one DB round-trip
regardless of how many handler types the previous process definition
declared.

Author: filiphr

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

@@ -0,0 +1,87 @@
+/* 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.flowable.engine.impl.bpmn.behavior;
+
+import org.flowable.bpmn.constants.BpmnXMLConstants;
+import org.flowable.common.engine.api.scope.ScopeTypes;
+import org.flowable.engine.impl.persistence.entity.ProcessDefinitionEntity;
+import org.flowable.engine.impl.util.CorrelationUtil;
+import org.flowable.engine.impl.util.CountingEntityUtil;
+import org.flowable.eventsubscription.api.EventSubscription;
+import org.flowable.eventsubscription.api.EventSubscriptionBuilder;
+import org.flowable.eventsubscription.service.EventSubscriptionService;
+
+/**
+ * Process-level event-registry start event behavior. Owns the deploy-time event-registry subscription
+ * registration for this start event, including the dynamic-correlation re-point logic when the
+ * process definition is superseded.
+ */
+public class EventRegistryStartEventActivityBehavior extends FlowNodeActivityBehavior implements ProcessLevelStartEventActivityBehavior {
+
+    private static final long serialVersionUID = 1L;
+
+    protected String eventDefinitionKey;
+    protected boolean manualCorrelation;
+
+    public EventRegistryStartEventActivityBehavior(String eventDefinitionKey, boolean manualCorrelation) {
+        this.eventDefinitionKey = eventDefinitionKey;
+        this.manualCorrelation = manualCorrelation;
+    }
+
+    @Override
+    public void deploy(ProcessLevelStartEventDeployContext context) {
+        // dynamic, manual subscription mode: deploy does not create a subscription — those are added
+        // explicitly by the application at runtime.
+        if (manualCorrelation) {
+            return;
+        }
+
+        ProcessDefinitionEntity processDefinition = context.getProcessDefinition();
+        EventSubscriptionService eventSubscriptionService = context.getEventSubscriptionService();
+        EventSubscriptionBuilder eventSubscriptionBuilder = eventSubscriptionService.createEventSubscriptionBuilder()
+                .eventType(eventDefinitionKey)
+                .activityId(context.getStartEvent().getId())
+                .processDefinitionId(processDefinition.getId())
+                .scopeType(ScopeTypes.BPMN)
+                .configuration(CorrelationUtil.getCorrelationKey(BpmnXMLConstants.ELEMENT_EVENT_CORRELATION_PARAMETER, context.getCommandContext(), context.getStartEvent(), null));
+
+        if (processDefinition.getTenantId() != null) {
+            eventSubscriptionBuilder.tenantId(processDefinition.getTenantId());
+        }
+
+        EventSubscription eventSubscription = eventSubscriptionBuilder.create();
+        CountingEntityUtil.handleInsertEventSubscriptionEntityCount(eventSubscription);
+    }
+
+    @Override
+    public void undeploy(ProcessLevelStartEventUndeployContext context) {
+        if (manualCorrelation) {
+            // dynamic mode: keep existing subscriptions but re-point them to the new process definition instead of deleting.
+            ProcessDefinitionEntity previousProcessDefinition = context.getPreviousProcessDefinition();
+            ProcessDefinitionEntity newProcessDefinition = context.getNewProcessDefinition();
+            context.getProcessEngineConfiguration().getEventSubscriptionServiceConfiguration().getEventSubscriptionService().updateEventSubscriptionProcessDefinitionId(
+                    previousProcessDefinition.getId(), newProcessDefinition.getId(),
+                    eventDefinitionKey, context.getStartEvent().getId(), newProcessDefinition.getKey(), null);
+        } else {
+            context.registerObsoleteEventSubscriptionType(eventDefinitionKey);
+        }
+    }
+
+    public String getEventDefinitionKey() {
+        return eventDefinitionKey;
+    }
+
+    public boolean isManualCorrelation() {
+        return manualCorrelation;
+    }
+}

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

@@ -0,0 +1,83 @@
+/* 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.flowable.engine.impl.bpmn.behavior;
+
+import java.util.List;
+
+import org.flowable.bpmn.model.MessageEventDefinition;
+import org.flowable.common.engine.api.FlowableException;
+import org.flowable.common.engine.impl.interceptor.CommandContext;
+import org.flowable.engine.impl.event.EventDefinitionExpressionUtil;
+import org.flowable.engine.impl.event.MessageEventHandler;
+import org.flowable.engine.impl.persistence.entity.ProcessDefinitionEntity;
+import org.flowable.engine.impl.util.CountingEntityUtil;
+import org.flowable.eventsubscription.service.EventSubscriptionService;
+import org.flowable.eventsubscription.service.impl.persistence.entity.EventSubscriptionEntity;
+import org.flowable.eventsubscription.service.impl.persistence.entity.MessageEventSubscriptionEntity;
+
+/**
+ * Process-level message start event behavior. Owns the deploy-time message subscription registration
+ * for this start event.
+ */
+public class MessageStartEventActivityBehavior extends FlowNodeActivityBehavior implements ProcessLevelStartEventActivityBehavior {
+
+    private static final long serialVersionUID = 1L;
+
+    protected MessageEventDefinition messageEventDefinition;
+
+    public MessageStartEventActivityBehavior(MessageEventDefinition messageEventDefinition) {
+        this.messageEventDefinition = messageEventDefinition;
+    }
+
+    @Override
+    public void deploy(ProcessLevelStartEventDeployContext context) {
+        CommandContext commandContext = context.getCommandContext();
+        EventSubscriptionService eventSubscriptionService = context.getEventSubscriptionService();
+        ProcessDefinitionEntity processDefinition = context.getProcessDefinition();
+
+        String messageName = EventDefinitionExpressionUtil.determineMessageName(commandContext, messageEventDefinition, processDefinition);
+        List<EventSubscriptionEntity> subscriptionsForSameMessageName = eventSubscriptionService
+                .findEventSubscriptionsByName(MessageEventHandler.EVENT_HANDLER_TYPE, messageName, processDefinition.getTenantId());
+
+        for (EventSubscriptionEntity eventSubscriptionEntity : subscriptionsForSameMessageName) {
+            // throw exception only if there's already a subscription as start event
+            if (eventSubscriptionEntity.getProcessInstanceId() == null || eventSubscriptionEntity.getProcessInstanceId().isEmpty()) {
+                // the event subscription has no instance-id, so it's a message start event
+                throw new FlowableException("Cannot deploy process definition '" + processDefinition.getResourceName()
+                        + "': there already is a message event subscription for the message with name '" + messageName + "'. For " + eventSubscriptionEntity);
+            }
+        }
+
+        MessageEventSubscriptionEntity newSubscription = eventSubscriptionService.createMessageEventSubscription();
+        newSubscription.setEventName(messageName);
+        newSubscription.setActivityId(context.getStartEvent().getId());
+        newSubscription.setConfiguration(processDefinition.getId());
+        newSubscription.setProcessDefinitionId(processDefinition.getId());
+
+        if (processDefinition.getTenantId() != null) {
+            newSubscription.setTenantId(processDefinition.getTenantId());
+        }
+
+        eventSubscriptionService.insertEventSubscription(newSubscription);
+        CountingEntityUtil.handleInsertEventSubscriptionEntityCount(newSubscription);
+    }
+
+    @Override
+    public void undeploy(ProcessLevelStartEventUndeployContext context) {
+        context.registerObsoleteEventSubscriptionType(MessageEventHandler.EVENT_HANDLER_TYPE);
+    }
+
+    public MessageEventDefinition getMessageEventDefinition() {
+        return messageEventDefinition;
+    }
+}

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

@@ -0,0 +1,43 @@
+/* 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.flowable.engine.impl.bpmn.behavior;
+
+/**
+ * Implemented by an {@code ActivityBehavior} attached to a process-level (i.e. NOT inside an
+ * {@code EventSubProcess}) {@code StartEvent} that registers a deploy-time artifact — an event
+ * subscription, a start timer job, etc. — and removes or updates that artifact when the process
+ * definition is superseded by a new version.
+ * <p>
+ * Both {@link #deploy} and {@link #undeploy} must be implemented; a behavior that opts into the
+ * deploy-time lifecycle owns both halves of it. Built-in start event types that don't register
+ * anything (e.g. none / error) simply don't implement this interface.
+ * <p>
+ * Custom integrations supplying their own {@code EventDefinition} + parse handler + behavior pick
+ * up deploy-time registration by implementing this interface — no change in
+ * {@code BpmnDeploymentHelper} is required.
+ */
+public interface ProcessLevelStartEventActivityBehavior {
+
+    /**
+     * Register the deploy-time artifact (event subscription, timer job, etc.) for this start event
+     * when its process definition is freshly deployed.
+     */
+    void deploy(ProcessLevelStartEventDeployContext context);
+
+    /**
+     * Remove or update the deploy-time artifact for this start event when its process definition is
+     * superseded by a new version. Called on the previous (now-superseded) process definition's
+     * start events.
+     */
+    void undeploy(ProcessLevelStartEventUndeployContext context);
+}

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

@@ -0,0 +1,75 @@
+/* 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.flowable.engine.impl.bpmn.behavior;
+
+import org.flowable.bpmn.model.BpmnModel;
+import org.flowable.bpmn.model.Process;
+import org.flowable.bpmn.model.StartEvent;
+import org.flowable.common.engine.impl.interceptor.CommandContext;
+import org.flowable.engine.impl.cfg.ProcessEngineConfigurationImpl;
+import org.flowable.engine.impl.persistence.entity.ProcessDefinitionEntity;
+import org.flowable.eventsubscription.service.EventSubscriptionService;
+
+/**
+ * Context passed to {@link ProcessLevelStartEventActivityBehavior#deploy} carrying the freshly-deployed
+ * process definition together with its parsed model and the start event in scope.
+ */
+public class ProcessLevelStartEventDeployContext {
+
+    protected final ProcessDefinitionEntity processDefinition;
+    protected final Process process;
+    protected final BpmnModel bpmnModel;
+    protected final StartEvent startEvent;
+    protected final ProcessEngineConfigurationImpl processEngineConfiguration;
+    protected final CommandContext commandContext;
+
+    public ProcessLevelStartEventDeployContext(ProcessDefinitionEntity processDefinition,
+            Process process, BpmnModel bpmnModel, StartEvent startEvent,
+            ProcessEngineConfigurationImpl processEngineConfiguration, CommandContext commandContext) {
+        this.processDefinition = processDefinition;
+        this.process = process;
+        this.bpmnModel = bpmnModel;
+        this.startEvent = startEvent;
+        this.processEngineConfiguration = processEngineConfiguration;
+        this.commandContext = commandContext;
+    }
+
+    public ProcessDefinitionEntity getProcessDefinition() {
+        return processDefinition;
+    }
+
+    public Process getProcess() {
+        return process;
+    }
+
+    public BpmnModel getBpmnModel() {
+        return bpmnModel;
+    }
+
+    public StartEvent getStartEvent() {
+        return startEvent;
+    }
+
+    public ProcessEngineConfigurationImpl getProcessEngineConfiguration() {
+        return processEngineConfiguration;
+    }
+
+    public CommandContext getCommandContext() {
+        return commandContext;
+    }
+
+    public EventSubscriptionService getEventSubscriptionService() {
+        return processEngineConfiguration.getEventSubscriptionServiceConfiguration().getEventSubscriptionService();
+    }
+
+}

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

@@ -0,0 +1,95 @@
+/* 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.flowable.engine.impl.bpmn.behavior;
+
+import java.util.Set;
+
+import org.flowable.bpmn.model.StartEvent;
+import org.flowable.common.engine.impl.interceptor.CommandContext;
+import org.flowable.engine.impl.cfg.ProcessEngineConfigurationImpl;
+import org.flowable.engine.impl.persistence.entity.ProcessDefinitionEntity;
+
+/**
+ * Context passed to {@link ProcessLevelStartEventActivityBehavior#undeploy} when a process definition
+ * is being superseded by a new version. Carries both the previous (now-superseded) process definition
+ * and the new one — most behaviors only need the previous process definition, but the EventRegistry "manual" re-point
+ * branch updates subscriptions to point at {@link #getNewProcessDefinition()}.
+ * <p>
+ * Behaviors register their obsolete event subscription / timer job handler types via
+ * {@link #registerObsoleteEventSubscriptionType(String)} and
+ * {@link #registerObsoleteTimerJobHandlerType(String)}. The deployer issues one mass-delete per
+ * unique registered type after the undeploy iteration — fewer DB round-trips than per-start-event
+ * deletes, and tighter than the historical fixed Message+Signal+EventRegistry sweep that always
+ * ran regardless of which types the previous process definition actually used.
+ */
+public class ProcessLevelStartEventUndeployContext {
+
+    protected final ProcessDefinitionEntity previousProcessDefinition;
+    protected final ProcessDefinitionEntity newProcessDefinition;
+    protected final StartEvent startEvent;
+    protected final ProcessEngineConfigurationImpl processEngineConfiguration;
+    protected final CommandContext commandContext;
+    protected final Set<String> obsoleteEventSubscriptionTypes;
+    protected final Set<String> obsoleteTimerJobHandlerTypes;
+
+    public ProcessLevelStartEventUndeployContext(ProcessDefinitionEntity previousProcessDefinition, ProcessDefinitionEntity newProcessDefinition,
+            StartEvent startEvent,
+            ProcessEngineConfigurationImpl processEngineConfiguration, CommandContext commandContext,
+            Set<String> obsoleteEventSubscriptionTypes, Set<String> obsoleteTimerJobHandlerTypes) {
+        this.previousProcessDefinition = previousProcessDefinition;
+        this.newProcessDefinition = newProcessDefinition;
+        this.startEvent = startEvent;
+        this.processEngineConfiguration = processEngineConfiguration;
+        this.commandContext = commandContext;
+        this.obsoleteEventSubscriptionTypes = obsoleteEventSubscriptionTypes;
+        this.obsoleteTimerJobHandlerTypes = obsoleteTimerJobHandlerTypes;
+    }
+
+    public ProcessDefinitionEntity getPreviousProcessDefinition() {
+        return previousProcessDefinition;
+    }
+
+    public ProcessDefinitionEntity getNewProcessDefinition() {
+        return newProcessDefinition;
+    }
+
+    public StartEvent getStartEvent() {
+        return startEvent;
+    }
+
+    public ProcessEngineConfigurationImpl getProcessEngineConfiguration() {
+        return processEngineConfiguration;
+    }
+
+    public CommandContext getCommandContext() {
+        return commandContext;
+    }
+
+    /**
+     * Registers an event subscription handler type that the deployer should mass-delete for the
+     * previous process definition after the undeploy pass. Multiple behaviors registering the same
+     * type result in a single DB sweep.
+     */
+    public void registerObsoleteEventSubscriptionType(String eventHandlerType) {
+        obsoleteEventSubscriptionTypes.add(eventHandlerType);
+    }
+
+    /**
+     * Registers a timer job handler type that the deployer should mass-cancel for the previous
+     * process definition after the undeploy pass. Multiple behaviors registering the same type
+     * result in a single DB sweep.
+     */
+    public void registerObsoleteTimerJobHandlerType(String timerJobHandlerType) {
+        obsoleteTimerJobHandlerTypes.add(timerJobHandlerType);
+    }
+}