Transaction handling refactoring (#771)

* unified new transaction handling in jpa repositories, extended Deployment Management and Action Repository

Signed-off-by: Bogdan Bondar <Bogdan.Bondar@bosch-si.com>

* moved Deployment Helper to utilities package

Signed-off-by: Bogdan Bondar <Bogdan.Bondar@bosch-si.com>

* removed superfluous utility method from Deployment Helper

Signed-off-by: Bogdan Bondar <Bogdan.Bondar@bosch-si.com>

* refactored distribution set to target assignment, fixed PR review findings

Signed-off-by: Bogdan Bondar <Bogdan.Bondar@bosch-si.com>

* fixed PR review findings

Signed-off-by: Bogdan Bondar <Bogdan.Bondar@bosch-si.com>

* added additional validation of active flag and current status fields for action status update repository method

Signed-off-by: Bogdan Bondar <Bogdan.Bondar@bosch-si.com>

* fixed timing issue in amqp message handler integration test, when validating target attributes update

Signed-off-by: Bogdan Bondar <Bogdan.Bondar@bosch-si.com>

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageHandlerServiceIntegrationTest.java

@@ -655,7 +655,7 @@ public class AmqpMessageHandlerServiceIntegrationTest extends AmqpServiceIntegra
         // update mode REPLACE
         updateAttributesWithUpdateModeReplace(controllerId);
 
-        // update mode REPLACE
+        // update mode MERGE
         updateAttributesWithUpdateModeMerge(controllerId);
 
         // update mode REMOVE
@@ -675,6 +675,7 @@ public class AmqpMessageHandlerServiceIntegrationTest extends AmqpServiceIntegra
         final Map<String, String> removeAttributes = new HashMap<>();
         removeAttributes.put("k1", "foo");
         removeAttributes.put("k3", "bar");
+
         final DmfAttributeUpdate remove = new DmfAttributeUpdate();
         remove.setMode(DmfUpdateMode.REMOVE);
         remove.getAttributes().putAll(removeAttributes);
@@ -694,6 +695,7 @@ public class AmqpMessageHandlerServiceIntegrationTest extends AmqpServiceIntegra
         final Map<String, String> mergeAttributes = new HashMap<>();
         mergeAttributes.put("k1", "v1_modified_again");
         mergeAttributes.put("k4", "v4");
+
         final DmfAttributeUpdate merge = new DmfAttributeUpdate();
         merge.setMode(DmfUpdateMode.MERGE);
         merge.getAttributes().putAll(mergeAttributes);
@@ -710,33 +712,33 @@ public class AmqpMessageHandlerServiceIntegrationTest extends AmqpServiceIntegra
     private void updateAttributesWithUpdateModeReplace(final String controllerId) {
 
         // send a update message with update mode REPLACE
-        final Map<String, String> replacementAttributes = new HashMap<>();
+        final Map<String, String> expectedAttributes = new HashMap<>();
-        replacementAttributes.put("k1", "v1_modified");
+        expectedAttributes.put("k1", "v1_modified");
-        replacementAttributes.put("k2", "v2");
+        expectedAttributes.put("k2", "v2");
-        replacementAttributes.put("k3", "v3");
+        expectedAttributes.put("k3", "v3");
+
         final DmfAttributeUpdate replace = new DmfAttributeUpdate();
         replace.setMode(DmfUpdateMode.REPLACE);
-        replace.getAttributes().putAll(replacementAttributes);
+        replace.getAttributes().putAll(expectedAttributes);
         sendUpdateAttributeMessage(controllerId, TENANT_EXIST, replace);
 
         // validate
-        final Map<String, String> expectedAttributes = replacementAttributes;
         assertUpdateAttributes(controllerId, expectedAttributes);
     }
 
     @Step
     private void updateAttributesWithoutUpdateMode(final String controllerId) {
 
-        // send a update message which does not specify a update mode
+        // send a update message which does not specify an update mode
-        final Map<String, String> initialAttributes = new HashMap<>();
+        final Map<String, String> expectedAttributes = new HashMap<>();
-        initialAttributes.put("k0", "v0");
+        expectedAttributes.put("k0", "v0");
-        initialAttributes.put("k1", "v1");
+        expectedAttributes.put("k1", "v1");
+
         final DmfAttributeUpdate defaultUpdate = new DmfAttributeUpdate();
-        defaultUpdate.getAttributes().putAll(initialAttributes);
+        defaultUpdate.getAttributes().putAll(expectedAttributes);
         sendUpdateAttributeMessage(controllerId, TENANT_EXIST, defaultUpdate);
 
         // validate
-        final Map<String, String> expectedAttributes = initialAttributes;
         assertUpdateAttributes(controllerId, expectedAttributes);
     }
 
@@ -807,7 +809,8 @@ public class AmqpMessageHandlerServiceIntegrationTest extends AmqpServiceIntegra
         verifyNumberOfDeadLetterMessages(3);
     }
 
-    private void sendUpdateAttributesMessageWithGivenAttributes(final String target, final String key, final String value) {
+    private void sendUpdateAttributesMessageWithGivenAttributes(final String target, final String key,
+            final String value) {
         final DmfAttributeUpdate controllerAttribute = new DmfAttributeUpdate();
         controllerAttribute.getAttributes().put(key, value);
         final Message message = createUpdateAttributesMessage(target, TENANT_EXIST, controllerAttribute);
@@ -893,7 +896,7 @@ public class AmqpMessageHandlerServiceIntegrationTest extends AmqpServiceIntegra
 
     private void verifyNumberOfDeadLetterMessages(final int numberOfInvocations) {
         assertEmptyReceiverQueueCount();
-        createConditionFactory()
+        createConditionFactory().until(() -> Mockito.verify(getDeadletterListener(), Mockito.times(numberOfInvocations))
-                .until(() -> Mockito.verify(getDeadletterListener(), Mockito.times(numberOfInvocations)).handleMessage(Mockito.any()));
+                .handleMessage(Mockito.any()));
     }
 }

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpServiceIntegrationTest.java

@@ -152,7 +152,7 @@ public abstract class AmqpServiceIntegrationTest extends AbstractAmqpIntegration
     protected void assertRequestAttributesUpdateMessage(final String target) {
         assertReplyMessageHeader(EventTopic.REQUEST_ATTRIBUTES_UPDATE, target);
     }
-    
+
     protected void assertRequestAttributesUpdateMessageAbsent() {
         assertThat(replyToListener.getEventTopicMessages()).doesNotContainKey(EventTopic.REQUEST_ATTRIBUTES_UPDATE);
     }
@@ -352,12 +352,18 @@ public abstract class AmqpServiceIntegrationTest extends AbstractAmqpIntegration
 
     @Step
     protected void assertUpdateAttributes(final String controllerId, final Map<String, String> attributes) {
-        final Target findByControllerId = waitUntilIsPresent(
+        waitUntilIsPresent(() -> controllerManagement.getByControllerId(controllerId));
-                () -> controllerManagement.getByControllerId(controllerId));
+
-        final Map<String, String> controllerAttributes = targetManagement
+        createConditionFactory().until(() -> {
-                .getControllerAttributes(findByControllerId.getControllerId());
+            try {
-        assertThat(controllerAttributes.size()).isEqualTo(attributes.size());
+                final Map<String, String> controllerAttributes = securityRule
-        attributes.forEach((k, v) -> assertKeyValueInMap(k, v, controllerAttributes));
+                        .runAsPrivileged(() -> targetManagement.getControllerAttributes(controllerId));
+                assertThat(controllerAttributes.size()).isEqualTo(attributes.size());
+                attributes.forEach((k, v) -> assertKeyValueInMap(k, v, controllerAttributes));
+            } catch (final Exception e) {
+                throw new RuntimeException(e);
+            }
+        });
     }
 
     private void assertKeyValueInMap(final String key, final String value,

hawkbit-repository/hawkbit-repository-api/src/main/java/org/eclipse/hawkbit/repository/DeploymentManagement.java

@@ -9,6 +9,7 @@
 package org.eclipse.hawkbit.repository;
 
 import java.util.Collection;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 
@@ -411,6 +412,17 @@ public interface DeploymentManagement {
     @PreAuthorize(SpringEvalExpressions.HAS_AUTH_UPDATE_TARGET)
     Action forceTargetAction(long actionId);
 
+    /**
+     * Sets the status of inactive scheduled {@link Action}s for the specified
+     * {@link Target}s to {@link Status#CANCELED}
+     *
+     * @param targetIds
+     *            ids of the {@link Target}s the actions belong to
+     * 
+     */
+    @PreAuthorize(SpringEvalExpressions.HAS_AUTH_UPDATE_TARGET)
+    void cancelInactiveScheduledActionsForTargets(List<Long> targetIds);
+
     /**
      * Starts all scheduled actions of an RolloutGroup parent.
      *

hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/repository/AbstractRolloutManagement.java

@@ -27,10 +27,6 @@ import org.springframework.integration.support.locks.LockRegistry;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.scheduling.annotation.AsyncResult;
 import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.TransactionDefinition;
-import org.springframework.transaction.support.DefaultTransactionDefinition;
-import org.springframework.transaction.support.TransactionCallback;
-import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.concurrent.ListenableFuture;
 
 /**
@@ -65,8 +61,8 @@ public abstract class AbstractRolloutManagement implements RolloutManagement {
             final DeploymentManagement deploymentManagement, final RolloutGroupManagement rolloutGroupManagement,
             final DistributionSetManagement distributionSetManagement, final ApplicationContext context,
             final ApplicationEventPublisher eventPublisher, final VirtualPropertyReplacer virtualPropertyReplacer,
-            final PlatformTransactionManager txManager, final TenantAware tenantAware,
+            final PlatformTransactionManager txManager, final TenantAware tenantAware, final LockRegistry lockRegistry,
-            final LockRegistry lockRegistry, final RolloutApprovalStrategy rolloutApprovalStrategy) {
+            final RolloutApprovalStrategy rolloutApprovalStrategy) {
         this.targetManagement = targetManagement;
         this.deploymentManagement = deploymentManagement;
         this.rolloutGroupManagement = rolloutGroupManagement;
@@ -80,14 +76,6 @@ public abstract class AbstractRolloutManagement implements RolloutManagement {
         this.rolloutApprovalStrategy = rolloutApprovalStrategy;
     }
 
-    protected Long runInNewTransaction(final String transactionName, final TransactionCallback<Long> action) {
-        final DefaultTransactionDefinition def = new DefaultTransactionDefinition();
-        def.setName(transactionName);
-        def.setReadOnly(false);
-        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
-        return new TransactionTemplate(txManager, def).execute(action);
-    }
-
     protected RolloutGroupsValidation validateTargetsInGroups(final List<RolloutGroup> groups, final String baseFilter,
             final long totalTargets) {
         final List<Long> groupTargetCounts = new ArrayList<>(groups.size());

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/ActionRepository.java

@@ -207,10 +207,50 @@ public interface ActionRepository extends BaseEntityRepository<JpaAction, Long>,
     void switchStatus(@Param("statusToSet") Action.Status statusToSet, @Param("targetsIds") List<Long> targetIds,
             @Param("active") boolean active, @Param("currentStatus") Action.Status currentStatus);
 
+    /**
+     *
+     * Retrieves all IDs for {@link Action}s referring to the given target IDs,
+     * active flag, current status and distribution set not requiring migration
+     * step.
+     *
+     * @param targetIds
+     *            the IDs of targets for the actions
+     * @param active
+     *            flag to indicate active/inactive actions
+     * @param currentStatus
+     *            the current status of the actions
+     * @return the found list of {@link Action} IDs
+     */
+    @Query("SELECT a.id FROM JpaAction a WHERE a.target IN :targetsIds AND a.active = :active AND a.status = :currentStatus AND a.distributionSet.requiredMigrationStep = false")
+    List<Long> findByTargetIdInAndIsActiveAndActionStatusAndDistributionSetNotRequiredMigrationStep(
+            @Param("targetsIds") List<Long> targetIds, @Param("active") boolean active,
+            @Param("currentStatus") Action.Status currentStatus);
+
+    /**
+     * Switches the status of actions from one specific status into another for
+     * given actions IDs, active flag and current status
+     *
+     * @param statusToSet
+     *            the new status the actions should get
+     * @param actionIds
+     *            the IDs of the actions which are affected
+     * @param active
+     *            flag to indicate active/inactive actions
+     * @param currentStatus
+     *            the current status of the actions
+     * @return the amount of updated actions
+     */
+    @Modifying
+    @Transactional
+    @Query("UPDATE JpaAction a SET a.status = :statusToSet WHERE a.id IN :actionIds AND a.active = :active AND a.status = :currentStatus")
+    int switchStatusForActionIdInAndIsActiveAndActionStatus(@Param("statusToSet") Action.Status statusToSet,
+            @Param("actionIds") List<Long> actionIds, @Param("active") boolean active,
+            @Param("currentStatus") Action.Status currentStatus);
+
     /**
      *
      * Retrieves all {@link Action}s which are active and referring to the given
-     * target Ids and distribution set required migration step.
+     * target Ids and distribution set not requiring migration step.
      *
      * @param targetIds
      *            the IDs of targets for the actions
@@ -226,7 +266,7 @@ public interface ActionRepository extends BaseEntityRepository<JpaAction, Long>,
     /**
      *
      * Retrieves all {@link Action}s which are active and referring to the given
-     * target Ids and distribution set required migration step.
+     * target Ids and distribution set not requiring migration step.
      *
      * @param targetIds
      *            the IDs of targets for the actions

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/JpaControllerManagement.java

@@ -60,6 +60,7 @@ import org.eclipse.hawkbit.repository.jpa.model.JpaDistributionSet;
 import org.eclipse.hawkbit.repository.jpa.model.JpaTarget;
 import org.eclipse.hawkbit.repository.jpa.model.JpaTarget_;
 import org.eclipse.hawkbit.repository.jpa.specifications.ActionSpecifications;
+import org.eclipse.hawkbit.repository.jpa.utils.DeploymentHelper;
 import org.eclipse.hawkbit.repository.jpa.utils.QuotaHelper;
 import org.eclipse.hawkbit.repository.model.Action;
 import org.eclipse.hawkbit.repository.model.Action.Status;
@@ -88,12 +89,9 @@ import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.retry.annotation.Retryable;
 import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.annotation.Isolation;
 import org.springframework.transaction.annotation.Transactional;
-import org.springframework.transaction.support.DefaultTransactionDefinition;
 import org.springframework.transaction.support.TransactionCallback;
-import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.validation.annotation.Validated;
 
 import com.google.common.base.Joiner;
@@ -178,14 +176,6 @@ public class JpaControllerManagement implements ControllerManagement {
         this.repositoryProperties = repositoryProperties;
     }
 
-    private <T> T runInNewTransaction(final String transactionName, final TransactionCallback<T> action) {
-        final DefaultTransactionDefinition def = new DefaultTransactionDefinition();
-        def.setName(transactionName);
-        def.setReadOnly(false);
-        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
-        return new TransactionTemplate(txManager, def).execute(action);
-    }
-
     @Override
     public String getPollingTime() {
         return systemSecurityContext.runAsSystem(() -> tenantConfigurationManagement
@@ -423,7 +413,8 @@ public class JpaControllerManagement implements ControllerManagement {
         try {
             events.stream().collect(Collectors.groupingBy(TargetPoll::getTenant)).forEach((tenant, polls) -> {
                 final TransactionCallback<Void> createTransaction = status -> updateLastTargetQueries(tenant, polls);
-                tenantAware.runAsTenant(tenant, () -> runInNewTransaction("flushUpdateQueue", createTransaction));
+                tenantAware.runAsTenant(tenant,
+                        () -> DeploymentHelper.runInNewTransaction(txManager, "flushUpdateQueue", createTransaction));
             });
         } catch (final RuntimeException ex) {
             LOG.error("Failed to persist UpdateQueue content.", ex);
@@ -605,8 +596,8 @@ public class JpaControllerManagement implements ControllerManagement {
 
         switch (actionStatus.getStatus()) {
         case ERROR:
-            final JpaTarget target = DeploymentHelper.updateTargetInfo((JpaTarget) action.getTarget(),
+            final JpaTarget target = (JpaTarget) action.getTarget();
-                    TargetUpdateStatus.ERROR, false);
+            target.setUpdateStatus(TargetUpdateStatus.ERROR);
             handleErrorOnAction(action, target);
             break;
         case FINISHED:
@@ -626,7 +617,7 @@ public class JpaControllerManagement implements ControllerManagement {
         if (controllerId != null) {
             targetManagement.requestControllerAttributes(controllerId);
         }
-        
+
         return savedAction;
     }
 
@@ -673,10 +664,10 @@ public class JpaControllerManagement implements ControllerManagement {
             final UpdateMode mode) {
 
         /*
-            Constraints on attribute keys & values are not validated by EclipseLink. Hence, they are validated here.
+         * Constraints on attribute keys & values are not validated by
+         * EclipseLink. Hence, they are validated here.
          */
-        if (data.entrySet().stream()
+        if (data.entrySet().stream().anyMatch(e -> !isAttributeEntryValid(e))) {
-                .anyMatch(e -> !isAttributeEntryValid(e))) {
             throw new InvalidTargetAttributeException();
         }
 

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/JpaDeploymentManagement.java

@@ -50,6 +50,7 @@ import org.eclipse.hawkbit.repository.jpa.model.JpaDistributionSet;
 import org.eclipse.hawkbit.repository.jpa.model.JpaTarget;
 import org.eclipse.hawkbit.repository.jpa.model.JpaTarget_;
 import org.eclipse.hawkbit.repository.jpa.rsql.RSQLUtility;
+import org.eclipse.hawkbit.repository.jpa.utils.DeploymentHelper;
 import org.eclipse.hawkbit.repository.jpa.utils.QuotaHelper;
 import org.eclipse.hawkbit.repository.model.Action;
 import org.eclipse.hawkbit.repository.model.Action.ActionType;
@@ -82,11 +83,8 @@ import org.springframework.orm.jpa.vendor.Database;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.retry.annotation.Retryable;
 import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.annotation.Isolation;
 import org.springframework.transaction.annotation.Transactional;
-import org.springframework.transaction.support.DefaultTransactionDefinition;
-import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.CollectionUtils;
 import org.springframework.validation.annotation.Validated;
 
@@ -142,7 +140,7 @@ public class JpaDeploymentManagement implements DeploymentManagement {
     private final TenantAware tenantAware;
     private final Database database;
 
-    JpaDeploymentManagement(final EntityManager entityManager, final ActionRepository actionRepository,
+    protected JpaDeploymentManagement(final EntityManager entityManager, final ActionRepository actionRepository,
             final DistributionSetRepository distributionSetRepository, final TargetRepository targetRepository,
             final ActionStatusRepository actionStatusRepository, final TargetManagement targetManagement,
             final AuditorAware<String> auditorProvider, final ApplicationEventPublisher eventPublisher,
@@ -253,102 +251,82 @@ public class JpaDeploymentManagement implements DeploymentManagement {
             final Collection<TargetWithActionType> targetsWithActionType, final String actionMessage,
             final AbstractDsAssignmentStrategy assignmentStrategy) {
 
-        final JpaDistributionSet set = distributionSetRepository.findOne(dsID);
+        final JpaDistributionSet distributionSetEntity = getAndValidateDsById(dsID);
-        if (set == null) {
+        final List<String> controllerIDs = getControllerIdsForAssignmentAndCheckQuota(targetsWithActionType,
-            throw new EntityNotFoundException(DistributionSet.class, dsID);
+                distributionSetEntity);
-        }
+        final List<JpaTarget> targetEntities = assignmentStrategy.findTargetsForAssignment(controllerIDs,
+                distributionSetEntity.getId());
 
-        if (!set.isComplete()) {
+        if (targetEntities.isEmpty()) {
-            throw new IncompleteDistributionSetException(
-                    "Distribution set of type " + set.getType().getKey() + " is incomplete: " + set.getId());
-        }
-
-        final List<String> controllerIDs = targetsWithActionType.stream().map(TargetWithActionType::getControllerId)
-                .collect(Collectors.toList());
-
-        // enforce the 'max targets per manual assignment' quota
-        if (!controllerIDs.isEmpty()) {
-            assertMaxTargetsPerManualAssignmentQuota(set.getId(), controllerIDs.size());
-        }
-
-        LOG.debug("assignDistribution({}) to {} targets", set, controllerIDs.size());
-
-        final Map<String, TargetWithActionType> targetsWithActionMap = targetsWithActionType.stream()
-                .collect(Collectors.toMap(TargetWithActionType::getControllerId, Function.identity()));
-
-        // split tIDs length into max entries in-statement because many database
-        // have constraint of max entries in in-statements e.g. Oracle with
-        // maximum 1000 elements, so we need to split the entries here and
-        // execute multiple statements we take the target only into account if
-        // the requested operation is no duplicate of a previous one
-        final List<JpaTarget> targets = assignmentStrategy.findTargetsForAssignment(controllerIDs, set.getId());
-
-        if (targets.isEmpty()) {
             // detaching as it is not necessary to persist the set itself
-            entityManager.detach(set);
+            entityManager.detach(distributionSetEntity);
             // return with nothing as all targets had the DS already assigned
             return new DistributionSetAssignmentResult(Collections.emptyList(), 0, targetsWithActionType.size(),
                     Collections.emptyList(), targetManagement);
         }
 
-        final List<List<Long>> targetIds = Lists.partition(
+        // split tIDs length into max entries in-statement because many database
-                targets.stream().map(Target::getId).collect(Collectors.toList()), Constants.MAX_ENTRIES_IN_STATEMENT);
+        // have constraint of max entries in in-statements e.g. Oracle with
+        // maximum 1000 elements, so we need to split the entries here and
+        // execute multiple statements
+        final List<List<Long>> targetEntitiesIdsChunks = Lists.partition(
+                targetEntities.stream().map(Target::getId).collect(Collectors.toList()),
+                Constants.MAX_ENTRIES_IN_STATEMENT);
 
         // override all active actions and set them into canceling state, we
         // need to remember which one we have been switched to canceling state
         // because for targets which we have changed to canceling we don't want
         // to publish the new action update event.
-        final Set<Long> targetIdsCancellList;
+        final Set<Long> cancelingTargetEntitiesIds = closeOrCancelActiveActions(assignmentStrategy,
-
+                targetEntitiesIdsChunks);
-        if (systemSecurityContext.runAsSystem(() -> tenantConfigurationManagement
-                .getConfigurationValue(TenantConfigurationKey.REPOSITORY_ACTIONS_AUTOCLOSE_ENABLED, Boolean.class)
-                .getValue())) {
-            targetIdsCancellList = Collections.emptySet();
-            assignmentStrategy.closeActiveActions(targetIds);
-        } else {
-            targetIdsCancellList = assignmentStrategy.cancelActiveActions(targetIds);
-        }
-
         // cancel all scheduled actions which are in-active, these actions were
         // not active before and the manual assignment which has been done
-        // cancels the
+        // cancels them
-        targetIds.forEach(tIds -> actionRepository.switchStatus(Status.CANCELED, tIds, false, Status.SCHEDULED));
+        targetEntitiesIdsChunks.forEach(this::cancelInactiveScheduledActionsForTargets);
 
-        // set assigned distribution set and TargetUpdateStatus
+        setAssignedDistributionSetAndTargetUpdateStatus(assignmentStrategy, distributionSetEntity,
-        final String currentUser;
+                targetEntitiesIdsChunks);
-        if (auditorProvider != null) {
-            currentUser = auditorProvider.getCurrentAuditor();
-        } else {
-            currentUser = null;
-        }
-
-        assignmentStrategy.updateTargetStatus(set, targetIds, currentUser);
-
-        final Map<String, JpaAction> targetIdsToActions = targets.stream()
-                .map(trg -> createTargetAction(assignmentStrategy, targetsWithActionType, controllerIDs, targets,
-                        targetsWithActionMap, trg, set))
-                .map(actionRepository::save)
-                .collect(Collectors.toMap(action -> action.getTarget().getControllerId(), Function.identity()));
 
+        final Map<String, JpaAction> controllerIdsToActions = createActions(targetsWithActionType, targetEntities,
+                assignmentStrategy, controllerIDs, distributionSetEntity);
         // create initial action status when action is created so we remember
         // the initial running status because we will change the status
         // of the action itself and with this action status we have a nicer
         // action history.
-        actionStatusRepository.save(targetIdsToActions.values().stream()
+        createActionsStatus(controllerIdsToActions.values(), assignmentStrategy, actionMessage);
-                .map(action -> assignmentStrategy.createActionStatus(action, actionMessage))
-                .collect(Collectors.toList()));
 
-        // detaching as it is not necessary to persist the set itself
+        detachEntitiesAndSendAssignmentEvents(distributionSetEntity, targetEntities, assignmentStrategy,
-        entityManager.detach(set);
+                cancelingTargetEntitiesIds, controllerIdsToActions);
-        // detaching as the entity has been updated by the JPQL query above
-        targets.forEach(entityManager::detach);
-
-        assignmentStrategy.sendAssignmentEvents(set, targets, targetIdsCancellList, targetIdsToActions);
 
         return new DistributionSetAssignmentResult(
-                targets.stream().map(Target::getControllerId).collect(Collectors.toList()), targets.size(),
+                targetEntities.stream().map(Target::getControllerId).collect(Collectors.toList()),
-                controllerIDs.size() - targets.size(), Lists.newArrayList(targetIdsToActions.values()),
+                targetEntities.size(), controllerIDs.size() - targetEntities.size(),
-                targetManagement);
+                Lists.newArrayList(controllerIdsToActions.values()), targetManagement);
+    }
+
+    private JpaDistributionSet getAndValidateDsById(final Long dsID) {
+        final JpaDistributionSet distributionSet = distributionSetRepository.findById(dsID)
+                .orElseThrow(() -> new EntityNotFoundException(DistributionSet.class, dsID));
+
+        if (!distributionSet.isComplete()) {
+            throw new IncompleteDistributionSetException("Distribution set of type "
+                    + distributionSet.getType().getKey() + " is incomplete: " + distributionSet.getId());
+        }
+
+        return distributionSet;
+    }
+
+    private List<String> getControllerIdsForAssignmentAndCheckQuota(
+            final Collection<TargetWithActionType> targetsWithActionType, final JpaDistributionSet distributionSet) {
+        final List<String> controllerIDs = targetsWithActionType.stream().map(TargetWithActionType::getControllerId)
+                .collect(Collectors.toList());
+
+        // enforce the 'max targets per manual assignment' quota
+        if (!controllerIDs.isEmpty()) {
+            assertMaxTargetsPerManualAssignmentQuota(distributionSet.getId(), controllerIDs.size());
+        }
+
+        return controllerIDs;
     }
 
     /**
@@ -360,9 +338,71 @@ public class JpaDeploymentManagement implements DeploymentManagement {
      * @param requested
      *            number of targets to check
      */
-    private void assertMaxTargetsPerManualAssignmentQuota(final Long id, final int requested) {
+    private void assertMaxTargetsPerManualAssignmentQuota(final Long distributionSetId,
-        QuotaHelper.assertAssignmentQuota(id, requested, quotaManagement.getMaxTargetsPerManualAssignment(),
+            final int requestedTargetsCount) {
-                Target.class, DistributionSet.class, null);
+        QuotaHelper.assertAssignmentQuota(distributionSetId, requestedTargetsCount,
+                quotaManagement.getMaxTargetsPerManualAssignment(), Target.class, DistributionSet.class, null);
+    }
+
+    private Set<Long> closeOrCancelActiveActions(final AbstractDsAssignmentStrategy assignmentStrategy,
+            final List<List<Long>> targetIdsChunks) {
+        if (isActionsAutocloseEnabled()) {
+            assignmentStrategy.closeActiveActions(targetIdsChunks);
+            return Collections.emptySet();
+        } else {
+            return assignmentStrategy.cancelActiveActions(targetIdsChunks);
+        }
+    }
+
+    protected boolean isActionsAutocloseEnabled() {
+        return systemSecurityContext.runAsSystem(() -> tenantConfigurationManagement
+                .getConfigurationValue(TenantConfigurationKey.REPOSITORY_ACTIONS_AUTOCLOSE_ENABLED, Boolean.class)
+                .getValue());
+    }
+
+    @Override
+    @Transactional(isolation = Isolation.READ_COMMITTED)
+    @Retryable(include = {
+            ConcurrencyFailureException.class }, maxAttempts = Constants.TX_RT_MAX, backoff = @Backoff(delay = Constants.TX_RT_DELAY))
+    public void cancelInactiveScheduledActionsForTargets(final List<Long> targetIds) {
+        actionRepository.switchStatus(Status.CANCELED, targetIds, false, Status.SCHEDULED);
+    }
+
+    private void setAssignedDistributionSetAndTargetUpdateStatus(final AbstractDsAssignmentStrategy assignmentStrategy,
+            final JpaDistributionSet set, final List<List<Long>> targetIdsChunks) {
+        final String currentUser = auditorProvider != null ? auditorProvider.getCurrentAuditor() : null;
+        assignmentStrategy.updateTargetStatus(set, targetIdsChunks, currentUser);
+    }
+
+    private Map<String, JpaAction> createActions(final Collection<TargetWithActionType> targetsWithActionType,
+            final List<JpaTarget> targets, final AbstractDsAssignmentStrategy assignmentStrategy,
+            final List<String> controllerIDs, final JpaDistributionSet set) {
+        final Map<String, TargetWithActionType> targetsWithActionMap = targetsWithActionType.stream()
+                .collect(Collectors.toMap(TargetWithActionType::getControllerId, Function.identity()));
+
+        return targets.stream()
+                .map(trg -> createTargetAction(assignmentStrategy, targetsWithActionType, controllerIDs, targets,
+                        targetsWithActionMap, trg, set))
+                .map(actionRepository::save)
+                .collect(Collectors.toMap(action -> action.getTarget().getControllerId(), Function.identity()));
+    }
+
+    private void createActionsStatus(final Collection<JpaAction> actions,
+            final AbstractDsAssignmentStrategy assignmentStrategy, final String actionMessage) {
+        actionStatusRepository
+                .save(actions.stream().map(action -> assignmentStrategy.createActionStatus(action, actionMessage))
+                        .collect(Collectors.toList()));
+    }
+
+    private void detachEntitiesAndSendAssignmentEvents(final JpaDistributionSet set, final List<JpaTarget> targets,
+            final AbstractDsAssignmentStrategy assignmentStrategy, final Set<Long> targetIdsCancellList,
+            final Map<String, JpaAction> controllerIdsToActions) {
+        // detaching as it is not necessary to persist the set itself
+        entityManager.detach(set);
+        // detaching as the entity has been updated by the JPQL query above
+        targets.forEach(entityManager::detach);
+
+        assignmentStrategy.sendAssignmentEvents(set, targets, targetIdsCancellList, controllerIdsToActions);
     }
 
     @Override
@@ -439,11 +479,7 @@ public class JpaDeploymentManagement implements DeploymentManagement {
 
     private long startScheduledActionsByRolloutGroupParentInNewTransaction(final Long rolloutId,
             final Long distributionSetId, final Long rolloutGroupParentId, final int limit) {
-        final DefaultTransactionDefinition def = new DefaultTransactionDefinition();
+        return DeploymentHelper.runInNewTransaction(txManager, "startScheduledActions-" + rolloutId, status -> {
-        def.setName("startScheduledActions-" + rolloutId);
-        def.setReadOnly(false);
-        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
-        return new TransactionTemplate(txManager, def).execute(status -> {
             final Page<Action> rolloutGroupActions = findActionsByRolloutAndRolloutGroupParent(rolloutId,
                     rolloutGroupParentId, limit);
 
@@ -797,4 +833,7 @@ public class JpaDeploymentManagement implements DeploymentManagement {
         }
     }
 
+    protected ActionRepository getActionRepository() {
+        return actionRepository;
+    }
 }

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/JpaRolloutManagement.java

@@ -53,6 +53,7 @@ import org.eclipse.hawkbit.repository.jpa.rollout.condition.RolloutGroupConditio
 import org.eclipse.hawkbit.repository.jpa.rsql.RSQLUtility;
 import org.eclipse.hawkbit.repository.jpa.specifications.RolloutSpecification;
 import org.eclipse.hawkbit.repository.jpa.specifications.SpecificationsBuilder;
+import org.eclipse.hawkbit.repository.jpa.utils.DeploymentHelper;
 import org.eclipse.hawkbit.repository.jpa.utils.QuotaHelper;
 import org.eclipse.hawkbit.repository.model.Action;
 import org.eclipse.hawkbit.repository.model.Action.ActionType;
@@ -158,7 +159,7 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
             final DistributionSetManagement distributionSetManagement, final ApplicationContext context,
             final ApplicationEventPublisher eventPublisher, final VirtualPropertyReplacer virtualPropertyReplacer,
             final PlatformTransactionManager txManager, final TenantAware tenantAware, final LockRegistry lockRegistry,
-            final Database database,  final RolloutApprovalStrategy rolloutApprovalStrategy) {
+            final Database database, final RolloutApprovalStrategy rolloutApprovalStrategy) {
         super(targetManagement, deploymentManagement, rolloutGroupManagement, distributionSetManagement, context,
                 eventPublisher, virtualPropertyReplacer, txManager, tenantAware, lockRegistry, rolloutApprovalStrategy);
         this.database = database;
@@ -361,7 +362,7 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
             if (!rolloutApprovalStrategy.isApprovalNeeded(rollout)) {
                 rollout.setStatus(RolloutStatus.READY);
                 LOGGER.debug("rollout {} creation done. Switch to READY.", rollout.getId());
-            } else  {
+            } else {
                 LOGGER.debug("rollout {} creation done. Switch to WAITING_FOR_APPROVAL.", rollout.getId());
                 rollout.setStatus(RolloutStatus.WAITING_FOR_APPROVAL);
                 rolloutApprovalStrategy.onApprovalRequired(rollout);
@@ -388,10 +389,12 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
         final List<Long> readyGroups = RolloutHelper.getGroupsByStatusIncludingGroup(rollout.getRolloutGroups(),
                 RolloutGroupStatus.READY, group);
 
-        final long targetsInGroupFilter = runInNewTransaction("countAllTargetsByTargetFilterQueryAndNotInRolloutGroups",
+        final long targetsInGroupFilter = DeploymentHelper.runInNewTransaction(txManager,
+                "countAllTargetsByTargetFilterQueryAndNotInRolloutGroups",
                 count -> targetManagement.countByRsqlAndNotInRolloutGroups(readyGroups, groupTargetFilter));
         final long expectedInGroup = Math.round(group.getTargetPercentage() / 100 * (double) targetsInGroupFilter);
-        final long currentlyInGroup = runInNewTransaction("countRolloutTargetGroupByRolloutGroup",
+        final long currentlyInGroup = DeploymentHelper.runInNewTransaction(txManager,
+                "countRolloutTargetGroupByRolloutGroup",
                 count -> rolloutTargetGroupRepository.countByRolloutGroup(group));
 
         // Switch the Group status to READY, when there are enough Targets in
@@ -413,8 +416,9 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
             } while (targetsLeftToAdd > 0);
 
             group.setStatus(RolloutGroupStatus.READY);
-            group.setTotalTargets(runInNewTransaction("countRolloutTargetGroupByRolloutGroup",
+            group.setTotalTargets(
-                    count -> rolloutTargetGroupRepository.countByRolloutGroup(group)).intValue());
+                    DeploymentHelper.runInNewTransaction(txManager, "countRolloutTargetGroupByRolloutGroup",
+                            count -> rolloutTargetGroupRepository.countByRolloutGroup(group)).intValue());
             return rolloutGroupRepository.save(group);
 
         } catch (final TransactionException e) {
@@ -426,7 +430,7 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
     private Long assignTargetsToGroupInNewTransaction(final JpaRollout rollout, final RolloutGroup group,
             final String targetFilter, final long limit) {
 
-        return runInNewTransaction("assignTargetsToRolloutGroup", status -> {
+        return DeploymentHelper.runInNewTransaction(txManager, "assignTargetsToRolloutGroup", status -> {
             final PageRequest pageRequest = new PageRequest(0, Math.toIntExact(limit));
             final List<Long> readyGroups = RolloutHelper.getGroupsByStatusIncludingGroup(rollout.getRolloutGroups(),
                     RolloutGroupStatus.READY, group);
@@ -475,14 +479,14 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
         final JpaRollout rollout = getRolloutAndThrowExceptionIfNotFound(rolloutId);
         RolloutHelper.verifyRolloutInStatus(rollout, RolloutStatus.WAITING_FOR_APPROVAL);
         switch (decision) {
-            case APPROVED:
+        case APPROVED:
-                rollout.setStatus(RolloutStatus.READY);
+            rollout.setStatus(RolloutStatus.READY);
-                break;
+            break;
-            case DENIED:
+        case DENIED:
-                rollout.setStatus(RolloutStatus.APPROVAL_DENIED);
+            rollout.setStatus(RolloutStatus.APPROVAL_DENIED);
-                break;
+            break;
-            default:
+        default:
-                throw new IllegalArgumentException("Unknown approval decision: " + decision);
+            throw new IllegalArgumentException("Unknown approval decision: " + decision);
         }
         rollout.setApprovalDecidedBy(rolloutApprovalStrategy.getApprovalUser(rollout));
         if (remark != null) {
@@ -577,7 +581,7 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
     }
 
     private Long createActionsForTargetsInNewTransaction(final long rolloutId, final long groupId, final int limit) {
-        return runInNewTransaction("createActionsForTargets", status -> {
+        return DeploymentHelper.runInNewTransaction(txManager, "createActionsForTargets", status -> {
             final PageRequest pageRequest = new PageRequest(0, limit);
             final Rollout rollout = rolloutRepository.findOne(rolloutId);
             final RolloutGroup group = rolloutGroupRepository.findOne(groupId);
@@ -608,7 +612,7 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
         // current scheduled action to cancel. E.g. a new scheduled action is
         // created.
         final List<Long> targetIds = targets.stream().map(Target::getId).collect(Collectors.toList());
-        actionRepository.switchStatus(Action.Status.CANCELED, targetIds, false, Action.Status.SCHEDULED);
+        deploymentManagement.cancelInactiveScheduledActionsForTargets(targetIds);
         targets.forEach(target -> {
 
             assertActionsPerTargetQuota(target, 1);
@@ -821,7 +825,7 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
         }
 
         try {
-            rollouts.forEach(rolloutId -> runInNewTransaction(handlerId + "-" + rolloutId,
+            rollouts.forEach(rolloutId -> DeploymentHelper.runInNewTransaction(txManager, handlerId + "-" + rolloutId,
                     status -> executeFittingHandler(rolloutId)));
         } finally {
             lock.unlock();

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/JpaSystemManagement.java

@@ -25,6 +25,7 @@ import org.eclipse.hawkbit.repository.jpa.configuration.MultiTenantJpaTransactio
 import org.eclipse.hawkbit.repository.jpa.model.JpaDistributionSetType;
 import org.eclipse.hawkbit.repository.jpa.model.JpaSoftwareModuleType;
 import org.eclipse.hawkbit.repository.jpa.model.JpaTenantMetaData;
+import org.eclipse.hawkbit.repository.jpa.utils.DeploymentHelper;
 import org.eclipse.hawkbit.repository.model.DistributionSetType;
 import org.eclipse.hawkbit.repository.model.SoftwareModuleType;
 import org.eclipse.hawkbit.repository.model.TenantMetaData;
@@ -46,11 +47,8 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.retry.annotation.Retryable;
 import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
-import org.springframework.transaction.support.DefaultTransactionDefinition;
-import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.validation.annotation.Validated;
 
 /**
@@ -207,12 +205,8 @@ public class JpaSystemManagement implements CurrentTenantCacheKeyGenerator, Syst
      * @return the initial created {@link TenantMetaData}
      */
     private TenantMetaData createInitialTenantMetaData(final String tenant) {
-        final DefaultTransactionDefinition def = new DefaultTransactionDefinition();
+        return systemSecurityContext.runAsSystemAsTenant(
-        def.setName("initial-tenant-creation");
+                () -> DeploymentHelper.runInNewTransaction(txManager, "initial-tenant-creation", status -> {
-        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
-        def.setReadOnly(false);
-        return systemSecurityContext
-                .runAsSystemAsTenant(() -> new TransactionTemplate(txManager, def).execute(status -> {
                     final DistributionSetType defaultDsType = createStandardSoftwareDataSetup();
                     return tenantMetaDataRepository.save(new JpaTenantMetaData(defaultDsType, tenant));
                 }), tenant);

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/utils/DeploymentHelper.java

@@ -6,18 +6,25 @@
  * which accompanies this distribution, and is available at
  * http://www.eclipse.org/legal/epl-v10.html
  */
-package org.eclipse.hawkbit.repository.jpa;
+package org.eclipse.hawkbit.repository.jpa.utils;
 
 import java.util.List;
 import java.util.stream.Collectors;
 
 import javax.validation.constraints.NotNull;
 
+import org.eclipse.hawkbit.repository.jpa.ActionRepository;
+import org.eclipse.hawkbit.repository.jpa.TargetRepository;
 import org.eclipse.hawkbit.repository.jpa.model.JpaAction;
 import org.eclipse.hawkbit.repository.jpa.model.JpaTarget;
 import org.eclipse.hawkbit.repository.model.Action;
 import org.eclipse.hawkbit.repository.model.Action.Status;
 import org.eclipse.hawkbit.repository.model.TargetUpdateStatus;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.DefaultTransactionDefinition;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
 
 /**
  * Utility class for deployment related topics.
@@ -29,31 +36,6 @@ public final class DeploymentHelper {
         // utility class
     }
 
-    /**
-     * Internal helper method used only inside service level. As a result is no
-     * additional security necessary.
-     *
-     * @param target
-     *            to update
-     * @param status
-     *            of the target
-     * @param setInstalledDate
-     *            to set
-     * @param targetInfoRepository
-     *            for the operation
-     *
-     * @return updated target
-     */
-    static JpaTarget updateTargetInfo(@NotNull final JpaTarget target, @NotNull final TargetUpdateStatus status,
-            final boolean setInstalledDate) {
-        target.setUpdateStatus(status);
-
-        if (setInstalledDate) {
-            target.setInstallationDate(System.currentTimeMillis());
-        }
-        return target;
-    }
-
     /**
      * This method is called, when cancellation has been successful. It sets the
      * action to canceled, resets the meta data of the target and in case there
@@ -65,10 +47,8 @@ public final class DeploymentHelper {
      *            for the operation
      * @param targetRepository
      *            for the operation
-     * @param targetInfoRepository
-     *            for the operation
      */
-    static void successCancellation(final JpaAction action, final ActionRepository actionRepository,
+    public static void successCancellation(final JpaAction action, final ActionRepository actionRepository,
             final TargetRepository targetRepository) {
 
         // set action inactive
@@ -81,11 +61,32 @@ public final class DeploymentHelper {
 
         if (nextActiveActions.isEmpty()) {
             target.setAssignedDistributionSet(target.getInstalledDistributionSet());
-            updateTargetInfo(target, TargetUpdateStatus.IN_SYNC, false);
+            target.setUpdateStatus(TargetUpdateStatus.IN_SYNC);
         } else {
             target.setAssignedDistributionSet(nextActiveActions.get(0).getDistributionSet());
         }
+
         targetRepository.save(target);
     }
 
+    /**
+     * Executes the modifying action in new transaction
+     *
+     * @param txManager
+     *            transaction manager interface
+     * @param transactionName
+     *            the name of the new transaction
+     * @param action
+     *            the callback to execute in new tranaction
+     *
+     * @return the result of the action
+     */
+    public static <T> T runInNewTransaction(@NotNull final PlatformTransactionManager txManager,
+            final String transactionName, @NotNull final TransactionCallback<T> action) {
+        final DefaultTransactionDefinition def = new DefaultTransactionDefinition();
+        def.setName(transactionName);
+        def.setReadOnly(false);
+        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+        return new TransactionTemplate(txManager, def).execute(action);
+    }
 }

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/utils/QuotaHelper.java

@@ -15,12 +15,10 @@ import javax.validation.constraints.NotNull;
 import org.eclipse.hawkbit.repository.exception.QuotaExceededException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.validation.annotation.Validated;
 
 /**
  * Helper class to check assignment quotas.
  */
-@Validated
 public final class QuotaHelper {
 
     /**