Fix channel handling in synchronization tasks

The channel URI was not consistently set in synchronization activities (import, recon, live sync, async update). It was hacked in a complex way, and still not reliable enough. This commit sets the task.channel property for such synchronization activities, hopefully providing channel to all places where it is needed. Some of the hacks could be thus removed. This resolves MID-7454.
Evolveum · Mar 24, 2022 · 9526b38 · 9526b38
1 parent a7c3cd3
commit 9526b38
Show file tree

Hide file tree

Showing 21 changed files with 107 additions and 58 deletions.
diff --git a/...-impl/src/main/java/com/evolveum/midpoint/model/impl/sync/SynchronizationServiceImpl.java b/...-impl/src/main/java/com/evolveum/midpoint/model/impl/sync/SynchronizationServiceImpl.java
@@ -909,6 +909,7 @@ private <F extends FocusType> void saveSyncMetadata(SynchronizationContext<F> sy
     private void savePendingDeltas(SynchronizationContext<?> syncCtx, OperationResult result) throws SchemaException {
         Task task = syncCtx.getTask();
         PrismObject<ShadowType> shadow = syncCtx.getShadowedResourceObject();
+        String channel = syncCtx.getChannel();
 
         try {
             beans.cacheRepositoryService.modifyObject(
@@ -917,9 +918,9 @@ private void savePendingDeltas(SynchronizationContext<?> syncCtx, OperationResul
                     syncCtx.getPendingShadowDeltas(),
                     result);
             syncCtx.clearPendingShadowDeltas();
-            task.recordObjectActionExecuted(shadow, ChangeType.MODIFY, null);
+            task.recordObjectActionExecuted(shadow, null, null, ChangeType.MODIFY, channel, null);
         } catch (ObjectNotFoundException ex) {
-            task.recordObjectActionExecuted(shadow, ChangeType.MODIFY, ex);
+            task.recordObjectActionExecuted(shadow, null, null, ChangeType.MODIFY, channel, ex);
             // This may happen e.g. during some recon-livesync interactions.
             // If the shadow is gone then it is gone. No point in recording the
             // situation any more.

diff --git a/...c/main/java/com/evolveum/midpoint/model/impl/sync/tasks/async/AsyncUpdateActivityRun.java b/...c/main/java/com/evolveum/midpoint/model/impl/sync/tasks/async/AsyncUpdateActivityRun.java
@@ -9,6 +9,7 @@
 
 import com.evolveum.midpoint.model.impl.sync.tasks.ResourceObjectClass;
 import com.evolveum.midpoint.repo.common.activity.run.*;
+import com.evolveum.midpoint.schema.constants.SchemaConstants;
 import com.evolveum.midpoint.xml.ns._public.common.common_3.*;
 
 import org.jetbrains.annotations.NotNull;
@@ -120,4 +121,9 @@ public boolean isExcludedFromStalenessChecking() {
         // This task does not have regularly updated progress. It cannot be watched for staleness (for now).
         return true;
     }
+
+    @Override
+    protected String getChannelOverride() {
+        return SchemaConstants.CHANNEL_ASYNC_UPDATE_URI;
+    }
 }
diff --git a/...impl/src/main/java/com/evolveum/midpoint/model/impl/sync/tasks/imp/ImportActivityRun.java b/...impl/src/main/java/com/evolveum/midpoint/model/impl/sync/tasks/imp/ImportActivityRun.java
@@ -107,4 +107,9 @@ public boolean processItem(@NotNull ShadowType object,
     private @NotNull ModelBeans getModelBeans() {
         return getActivityHandler().getModelBeans();
     }
+
+    @Override
+    protected String getChannelOverride() {
+        return SchemaConstants.CHANNEL_IMPORT_URI;
+    }
 }
diff --git a/...n/java/com/evolveum/midpoint/model/impl/sync/tasks/recon/RemainingShadowsActivityRun.java b/...n/java/com/evolveum/midpoint/model/impl/sync/tasks/recon/RemainingShadowsActivityRun.java
@@ -257,4 +257,9 @@ private PrismObject<ShadowType> reloadShadow(ShadowType originalShadow, Task tas
     public long getShadowReconCount() {
         return transientRunStatistics.getItemsProcessed();
     }
+
+    @Override
+    protected String getChannelOverride() {
+        return SchemaConstants.CHANNEL_RECON_URI;
+    }
 }
diff --git a/...olveum/midpoint/model/impl/sync/tasks/recon/ResourceObjectsReconciliationActivityRun.java b/...olveum/midpoint/model/impl/sync/tasks/recon/ResourceObjectsReconciliationActivityRun.java
@@ -100,4 +100,9 @@ public long getResourceReconCount() {
     public long getResourceReconErrors() {
         return transientRunStatistics.getErrors();
     }
+
+    @Override
+    protected String getChannelOverride() {
+        return SchemaConstants.CHANNEL_RECON_URI;
+    }
 }
diff --git a/...l/src/main/java/com/evolveum/midpoint/model/impl/sync/tasks/sync/LiveSyncActivityRun.java b/...l/src/main/java/com/evolveum/midpoint/model/impl/sync/tasks/sync/LiveSyncActivityRun.java
@@ -177,4 +177,9 @@ public boolean processItem(@NotNull ItemProcessingRequest<LiveSyncEvent> request
                 getRunningTask().getExtensionPropertyRealValue(SchemaConstants.MODEL_EXTENSION_RETRY_LIVE_SYNC_ERRORS));
         return retryErrors ? STOP : CONTINUE;
     }
+
+    @Override
+    protected String getChannelOverride() {
+        return SchemaConstants.CHANNEL_LIVE_SYNC_URI;
+    }
 }
diff --git a/...oning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/ProvisioningContext.java b/...oning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/ProvisioningContext.java
@@ -107,13 +107,6 @@ public class ProvisioningContext {
      */
     private Collection<ResourceObjectPattern> protectedObjectPatterns;
 
-    /**
-     * Override for values returned by {@link #getChannel()} method.
-     *
-     * TODO Remove this ugly hack. The channel should be read solely from the task.
-     */
-    private String channelOverride;
-
     /** Creating context from scratch. */
     ProvisioningContext(
             @NotNull Task task,
@@ -138,7 +131,6 @@ public class ProvisioningContext {
         this.contextFactory = originalCtx.contextFactory;
         this.connectorMap.putAll(originalCtx.connectorMap);
         this.resourceSchema = originalCtx.resourceSchema;
-        this.channelOverride = originalCtx.channelOverride;
         this.getOperationOptions = originalCtx.getOperationOptions; // OK?
         this.propagation = originalCtx.propagation;
         // Not copying protected account patters because these are object type specific.
@@ -293,12 +285,7 @@ public Collection<ResourceObjectPattern> getProtectedAccountPatterns(
     }
 
     public String getChannel() {
-        Task task = getTask();
-        if (channelOverride == null) {
-            return task.getChannel();
-        } else {
-            return channelOverride;
-        }
+        return task.getChannel();
     }
 
     public <T extends CapabilityType> ConnectorInstance getConnector(Class<T> operationCapabilityClass, OperationResult result)
@@ -469,10 +456,6 @@ public CachingStategyType getCachingStrategy()
         return ProvisioningUtil.getCachingStrategy(this);
     }
 
-    public void setChannelOverride(String channelOverride) {
-        this.channelOverride = channelOverride;
-    }
-
     public String toHumanReadableDescription() {
         if (resourceObjectDefinition != null) {
             return resourceObjectDefinition.getHumanReadableName() + " @" + resource;
@@ -481,15 +464,6 @@ public String toHumanReadableDescription() {
         }
     }
 
-    public Object toHumanReadableDescriptionLazy() {
-        return new Object() {
-            @Override
-            public String toString() {
-                return toHumanReadableDescription();
-            }
-        };
-    }
-
     public boolean isInMaintenance() {
         return ResourceTypeUtil.isInMaintenance(resource);
     }

diff --git a/...sioning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/AddHelper.java b/...sioning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/AddHelper.java
@@ -360,11 +360,10 @@ void notifyAfterAdd(
             PrismObject<ShadowType> addedShadow,
             ProvisioningOperationState<AsynchronousOperationReturnValue<PrismObject<ShadowType>>> opState,
             Task task,
-            OperationResult parentResult)
-            throws ObjectNotFoundException, SchemaException, CommunicationException, ConfigurationException, ExpressionEvaluationException {
+            OperationResult parentResult) {
         ObjectDelta<ShadowType> delta = DeltaFactory.Object.createAddDelta(addedShadow);
-        ResourceOperationDescription operationDescription = Util.createSuccessOperationDescription(ctx, addedShadow,
-                delta, parentResult);
+        ResourceOperationDescription operationDescription = Util.createSuccessOperationDescription(
+                ctx, addedShadow, delta, parentResult);
 
         if (opState.isExecuting()) {
             eventDispatcher.notifyInProgress(operationDescription, task, parentResult);

diff --git a/...ning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/DeleteHelper.java b/...ning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/DeleteHelper.java
@@ -250,8 +250,7 @@ void notifyAfterDelete(
             PrismObject<ShadowType> shadow,
             ProvisioningOperationState<AsynchronousOperationResult> opState,
             Task task,
-            OperationResult parentResult)
-            throws ObjectNotFoundException, SchemaException, CommunicationException, ConfigurationException, ExpressionEvaluationException {
+            OperationResult parentResult) {
         ObjectDelta<ShadowType> delta = prismContext.deltaFactory().object().createDeleteDelta(shadow.getCompileTimeClass(),
                 shadow.getOid());
         ResourceOperationDescription operationDescription = createSuccessOperationDescription(ctx, shadow,

diff --git a/...ning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ModifyHelper.java b/...ning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ModifyHelper.java
@@ -263,13 +263,12 @@ void notifyAfterModify(
             Collection<? extends ItemDelta> modifications,
             ProvisioningOperationState<AsynchronousOperationReturnValue<Collection<PropertyDelta<PrismPropertyValue>>>> opState,
             Task task,
-            OperationResult parentResult)
-            throws ObjectNotFoundException, SchemaException, CommunicationException, ConfigurationException, ExpressionEvaluationException {
+            OperationResult parentResult) {
 
         ObjectDelta<ShadowType> delta = prismContext.deltaFactory().object().createModifyDelta(repoShadow.getOid(), modifications,
                 repoShadow.getCompileTimeClass());
-        ResourceOperationDescription operationDescription = createSuccessOperationDescription(ctx, repoShadow,
-                delta, parentResult);
+        ResourceOperationDescription operationDescription =
+                createSuccessOperationDescription(ctx, repoShadow, delta, parentResult);
 
         if (opState.isExecuting()) {
             eventDispatcher.notifyInProgress(operationDescription, task, parentResult);

diff --git a/...ing-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/RefreshHelper.java b/...ing-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/RefreshHelper.java
@@ -327,8 +327,8 @@ private void refreshShadowAsyncStatus(ProvisioningContext ctx, PrismObject<Shado
         }
 
         for (ObjectDelta<ShadowType> notificationDelta: notificationDeltas) {
-            ResourceOperationDescription operationDescription = createSuccessOperationDescription(ctx, repoShadow,
-                    notificationDelta, parentResult);
+            ResourceOperationDescription operationDescription =
+                    createSuccessOperationDescription(ctx, repoShadow, notificationDelta, parentResult);
             operationListener.notifySuccess(operationDescription, task, parentResult);
         }
 
@@ -536,8 +536,8 @@ private PrismObject<ShadowType> deleteDeadShadowIfPossible(ProvisioningContext c
             LOGGER.debug("Deleting dead {} because it is expired", repoShadow);
             shadowManager.deleteShadow(repoShadow, task, parentResult);
             definitionsHelper.applyDefinition(repoShadow, task, parentResult);
-            ResourceOperationDescription operationDescription = createSuccessOperationDescription(ctx, repoShadow,
-                    repoShadow.createDeleteDelta(), parentResult);
+            ResourceOperationDescription operationDescription =
+                    createSuccessOperationDescription(ctx, repoShadow, repoShadow.createDeleteDelta(), parentResult);
             operationListener.notifySuccess(operationDescription, task, parentResult);
             return null;
         }

diff --git a/...pl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ShadowedAsyncChange.java b/...pl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ShadowedAsyncChange.java
@@ -11,6 +11,7 @@
 
 import com.evolveum.midpoint.schema.AcknowledgementSink;
 
+import com.evolveum.midpoint.schema.constants.SchemaConstants;
 import com.evolveum.midpoint.schema.result.OperationResult;
 
 import org.jetbrains.annotations.NotNull;
@@ -32,4 +33,9 @@ public ShadowedAsyncChange(@NotNull ResourceObjectAsyncChange resourceObjectChan
     public void acknowledge(boolean release, OperationResult result) {
         resourceObjectChange.acknowledge(release, result);
     }
+
+    @Override
+    protected String getDefaultChannel() {
+        return SchemaConstants.CHANNEL_ASYNC_UPDATE_URI;
+    }
 }
diff --git a/...ng-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ShadowedChange.java b/...ng-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ShadowedChange.java
@@ -58,7 +58,7 @@
  * This means that it is connected to repository shadow, and this shadow is updated
  * with the appropriate information.
  */
-public class ShadowedChange<ROC extends ResourceObjectChange> implements InitializableMixin {
+public abstract class ShadowedChange<ROC extends ResourceObjectChange> implements InitializableMixin {
 
     private static final Trace LOGGER = TraceManager.getTrace(ShadowedChange.class);
 
@@ -385,9 +385,15 @@ private void markIndexOnlyItemsAsIncomplete(PrismObject<ShadowType> resourceObje
     }
 
     private String getChannel() {
-        return ObjectUtils.defaultIfNull(context.getChannel(), SchemaConstants.CHANNEL_LIVE_SYNC_URI);
+        return ObjectUtils.defaultIfNull(context.getChannel(), getDefaultChannel());
     }
 
+    /**
+     * Default channel for given change. The usefulness of this method is questionable,
+     * as the context should have the correct channel already set.
+     */
+    protected abstract String getDefaultChannel();
+
     public Collection<ResourceAttribute<?>> getIdentifiers() {
         return resourceObjectChange.getIdentifiers();
     }

diff --git a/...src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ShadowedExternalChange.java b/...src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ShadowedExternalChange.java
@@ -20,4 +20,9 @@ public class ShadowedExternalChange extends ShadowedChange<ExternalResourceObjec
     public ShadowedExternalChange(@NotNull ExternalResourceObjectChange resourceObjectChange, ChangeProcessingBeans beans) {
         super(resourceObjectChange, beans);
     }
+
+    @Override
+    protected String getDefaultChannel() {
+        return null; // TODO ?
+    }
 }
diff --git a/...src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ShadowedLiveSyncChange.java b/...src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/ShadowedLiveSyncChange.java
@@ -10,6 +10,8 @@
 import com.evolveum.midpoint.provisioning.api.LiveSyncToken;
 import com.evolveum.midpoint.schema.AcknowledgementSink;
 
+import com.evolveum.midpoint.schema.constants.SchemaConstants;
+
 import org.jetbrains.annotations.NotNull;
 
 import com.evolveum.midpoint.provisioning.impl.resourceobjects.ResourceObjectLiveSyncChange;
@@ -27,4 +29,9 @@ public ShadowedLiveSyncChange(@NotNull ResourceObjectLiveSyncChange resourceObje
     public LiveSyncToken getToken() {
         return resourceObjectChange.getToken();
     }
+
+    @Override
+    protected String getDefaultChannel() {
+        return SchemaConstants.CHANNEL_LIVE_SYNC_URI;
+    }
 }
diff --git a/...provisioning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/Util.java b/...provisioning-impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/Util.java
@@ -64,9 +64,7 @@ static boolean shouldExecuteResourceOperationDirectly(ProvisioningContext ctx) t
     }
 
     static ResourceOperationDescription createSuccessOperationDescription(ProvisioningContext ctx,
-            PrismObject<ShadowType> shadowType, ObjectDelta<? extends ShadowType> delta, OperationResult parentResult)
-                    throws ObjectNotFoundException, SchemaException, CommunicationException,
-                    ConfigurationException, ExpressionEvaluationException {
+            PrismObject<ShadowType> shadowType, ObjectDelta<? extends ShadowType> delta, OperationResult parentResult) {
         ResourceOperationDescription operationDescription = new ResourceOperationDescription();
         operationDescription.setCurrentShadow(shadowType);
         operationDescription.setResource(ctx.getResource().asPrismObject());

diff --git a/...impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/sync/AsyncUpdater.java b/...impl/src/main/java/com/evolveum/midpoint/provisioning/impl/shadows/sync/AsyncUpdater.java
@@ -51,10 +51,6 @@ public void processAsynchronousUpdates(ResourceShadowCoordinates shadowCoordinat
 
         ProvisioningContext globalContext = ctxFactory.createForCoordinates(shadowCoordinates, callerTask, callerResult);
 
-        // This is a bit of hack to propagate information about async update channel to upper layers
-        // e.g. to implement MID-5853. TODO fix this hack
-        globalContext.setChannelOverride(SchemaConstants.CHANNEL_ASYNC_UPDATE_URI);
-
         IndividualEventsAcknowledgeGate<AsyncUpdateEvent> acknowledgeGate = new IndividualEventsAcknowledgeGate<>();
 
         ResourceObjectAsyncChangeListener listener = (resourceObjectChange, lTask, lResult) -> {

diff --git a/...ain/java/com/evolveum/midpoint/provisioning/ucf/api/async/AsyncUpdateMessageListener.java b/...ain/java/com/evolveum/midpoint/provisioning/ucf/api/async/AsyncUpdateMessageListener.java
@@ -10,12 +10,9 @@
 import com.evolveum.midpoint.schema.AcknowledgementSink;
 import com.evolveum.midpoint.xml.ns._public.common.common_3.AsyncUpdateMessageType;
 
-import com.google.common.annotations.VisibleForTesting;
-
 /**
  * Listener that receives "raw" async update messages from asynchronous update source (e.g. AMQP, JMS, REST, ...).
  */
-@VisibleForTesting // just to provide mock implementations
 public interface AsyncUpdateMessageListener {
 
     /**

diff --git a/...on/src/main/java/com/evolveum/midpoint/repo/common/activity/run/IterativeActivityRun.java b/...on/src/main/java/com/evolveum/midpoint/repo/common/activity/run/IterativeActivityRun.java
@@ -14,7 +14,6 @@
 
 import java.util.Objects;
 
-import com.evolveum.midpoint.prism.PrismContext;
 import com.evolveum.midpoint.prism.xml.XmlTypeConverter;
 
 import com.evolveum.midpoint.repo.common.activity.run.processing.ItemProcessingRequest;
@@ -175,8 +174,10 @@ public IterativeActivityRun(@NotNull ActivityRunInstantiationContext<WD, AH> con
 
         LOGGER.trace("{}: Starting with local coordinator task {}", shortName, getRunningTask());
 
+        String originalChannel = getRunningTask().getChannel();
         try {
             enableGlobalConnIdOperationsListener();
+            overrideTaskChannelIfNeeded();
 
             transientRunStatistics.recordRunStart(getStartTimestampRequired());
 
@@ -196,10 +197,27 @@ public IterativeActivityRun(@NotNull ActivityRunInstantiationContext<WD, AH> con
 
         } finally {
             disableGlobalConnIdOperationsListener();
+            cancelTaskChannelOverride(originalChannel);
             getActivityState().getConnIdOperationsReport().flush(getRunningTask(), result);
         }
     }
 
+    private void cancelTaskChannelOverride(String originalChannel) {
+        getRunningTask().setChannel(originalChannel);
+    }
+
+    private void overrideTaskChannelIfNeeded() {
+        String channelOverride = getChannelOverride();
+        if (channelOverride != null) {
+            getRunningTask().setChannel(channelOverride);
+        }
+    }
+
+    /** Channel URI that should be set into the task during this activity run. (If not null.) */
+    protected @Nullable String getChannelOverride() {
+        return null;
+    }
+
     /**
      * Bucketed version of the run.
      */
@@ -744,7 +762,7 @@ public final ConnIdOperationsReport getConnIdOperationsReport() {
     private void reportBucketCompleted(BucketProcessingRecord processingRecord, OperationResult result) {
         if (shouldReportBuckets()) {
             activityState.getBucketsReport().recordBucketCompleted(
-                    new BucketProcessingRecordType(PrismContext.get())
+                    new BucketProcessingRecordType()
                             .sequentialNumber(bucket.getSequentialNumber())
                             .content(bucket.getContent())
                             .size(processingRecord.getTotalSize())
@@ -761,7 +779,7 @@ private void reportBucketCompleted(BucketProcessingRecord processingRecord, Oper
     private void reportBucketAnalyzed(Integer size, OperationResult result) {
         if (shouldReportBuckets()) {
             activityState.getBucketsReport().recordBucketCompleted(
-                    new BucketProcessingRecordType(PrismContext.get())
+                    new BucketProcessingRecordType()
                             .sequentialNumber(bucket.getSequentialNumber())
                             .content(bucket.getContent())
                             .size(size),

diff --git a/...il/src/main/java/com/evolveum/midpoint/test/asserter/ActionsExecutedPartInfoAsserter.java b/...il/src/main/java/com/evolveum/midpoint/test/asserter/ActionsExecutedPartInfoAsserter.java
@@ -141,5 +141,14 @@ public ActionsExecutedPartInfoAsserter<RA> assertEmpty() {
         return this;
     }
 
+    public ActionsExecutedPartInfoAsserter<RA> assertChannels(String... uris) {
+        assertThat(getAllChannels()).as("all channels").containsExactlyInAnyOrder(uris);
+        return this;
+    }
 
+    public Collection<String> getAllChannels() {
+        return getEntries().stream()
+                .map(ObjectActionsExecutedEntryType::getChannel)
+                .collect(Collectors.toSet());
+    }
 }