feat(jobs): added JobStepProperty overrides on JobEngineStartOptions

Signed-off-by: Alberto Codutti <alberto.codutti@eurotech.com>
eclipse · Jul 3, 2023 · 8729304 · 8729304
1 parent 9d55c15
commit 8729304
Show file tree

Hide file tree

Showing 16 changed files with 225 additions and 7 deletions.
diff --git a/broker/core/pom.xml b/broker/core/pom.xml
@@ -208,6 +208,11 @@
             <artifactId>log4j-to-slf4j</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.kapua</groupId>
+            <artifactId>kapua-job-internal</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.eclipse.kapua</groupId>
             <artifactId>kapua-qa-markers</artifactId>

diff --git a/broker/core/src/test/resources/locator.xml b/broker/core/src/test/resources/locator.xml
@@ -76,6 +76,9 @@
         <api>org.eclipse.kapua.service.device.registry.event.DeviceEventService</api>
         <api>org.eclipse.kapua.service.device.registry.lifecycle.DeviceLifeCycleService</api>
 
+        <api>org.eclipse.kapua.service.job.step.JobStepService</api>
+        <api>org.eclipse.kapua.service.job.step.JobStepFactory</api>
+
         <api>org.eclipse.kapua.service.tag.TagFactory</api>
         <api>org.eclipse.kapua.service.tag.TagService</api>
 

diff --git a/job-engine/api/src/main/java/org/eclipse/kapua/job/engine/JobStartOptions.java b/job-engine/api/src/main/java/org/eclipse/kapua/job/engine/JobStartOptions.java
@@ -15,6 +15,7 @@
 import org.eclipse.kapua.KapuaSerializable;
 import org.eclipse.kapua.model.id.KapuaId;
 import org.eclipse.kapua.model.id.KapuaIdAdapter;
+import org.eclipse.kapua.service.job.step.definition.JobStepProperty;
 import org.eclipse.kapua.service.job.targets.JobTarget;
 
 import javax.xml.bind.annotation.XmlAccessType;
@@ -25,6 +26,7 @@
 import javax.xml.bind.annotation.XmlTransient;
 import javax.xml.bind.annotation.XmlType;
 import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter;
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -74,6 +76,12 @@ public interface JobStartOptions extends KapuaSerializable {
     @XmlTransient
     void addTargetIdToSublist(KapuaId targetId);
 
+    List<JobStepProperty> getStepPropertiesOverrides();
+
+    void addStepPropertyOverride(JobStepProperty jobStepPropertyOverride);
+
+    void setStepPropertiesOverrides(List<JobStepProperty> jobStepPropertiesOverrides);
+
     /**
      * Gets whether or not the {@link JobTarget#getStepIndex()} needs to be reset to the given {@link #getFromStepIndex()}.
      *
@@ -121,4 +129,5 @@ public interface JobStartOptions extends KapuaSerializable {
      * @since 1.1.0
      */
     void setEnqueue(boolean enqueue);
+
 }
diff --git a/job-engine/app/core/pom.xml b/job-engine/app/core/pom.xml
@@ -35,6 +35,10 @@
             <groupId>org.eclipse.kapua</groupId>
             <artifactId>kapua-job-engine-jbatch</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.kapua</groupId>
+            <artifactId>kapua-job-internal</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.eclipse.kapua</groupId>
             <artifactId>kapua-rest-api-core</artifactId>

diff --git a/...ore/src/main/java/org/eclipse/kapua/job/engine/app/core/jackson/ObjectMapperProvider.java b/...ore/src/main/java/org/eclipse/kapua/job/engine/app/core/jackson/ObjectMapperProvider.java
@@ -19,8 +19,10 @@
 import org.eclipse.kapua.job.engine.JobStartOptions;
 import org.eclipse.kapua.job.engine.app.core.jackson.mixin.IsJobRunningResponseMixin;
 import org.eclipse.kapua.job.engine.app.core.jackson.mixin.JobStartOptionsMixin;
+import org.eclipse.kapua.job.engine.app.core.jackson.mixin.JobStepPropertyMixin;
 import org.eclipse.kapua.job.engine.app.core.jackson.mixin.KapuaIdMixin;
 import org.eclipse.kapua.model.id.KapuaId;
+import org.eclipse.kapua.service.job.step.definition.JobStepProperty;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -33,6 +35,7 @@ public ObjectMapperProvider() {
         objectMapper = new ObjectMapper();
         objectMapper.addMixIn(KapuaId.class, KapuaIdMixin.class);
         objectMapper.addMixIn(JobStartOptions.class, JobStartOptionsMixin.class);
+        objectMapper.addMixIn(JobStepProperty.class, JobStepPropertyMixin.class);
         objectMapper.addMixIn(IsJobRunningResponse.class, IsJobRunningResponseMixin.class);
     }
 

diff --git a/...c/main/java/org/eclipse/kapua/job/engine/app/core/jackson/mixin/JobStepPropertyMixin.java b/...c/main/java/org/eclipse/kapua/job/engine/app/core/jackson/mixin/JobStepPropertyMixin.java
@@ -0,0 +1,22 @@
+/*******************************************************************************
+ * Copyright (c) 2021, 2022 Eurotech and/or its affiliates and others
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *
+ * Contributors:
+ *     Eurotech - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.kapua.job.engine.app.core.jackson.mixin;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.eclipse.kapua.service.job.step.definition.internal.JobStepPropertyImpl;
+
+@JsonDeserialize(as = JobStepPropertyImpl.class)
+@JsonIgnoreProperties("type")
+public interface JobStepPropertyMixin {
+}
diff --git a/.../app/web/src/main/java/org/eclipse/kapua/job/engine/app/web/jaxb/JaxbContextResolver.java b/.../app/web/src/main/java/org/eclipse/kapua/job/engine/app/web/jaxb/JaxbContextResolver.java
@@ -30,6 +30,8 @@
 import org.eclipse.kapua.app.api.core.exception.model.JobStoppingExceptionInfo;
 import org.eclipse.kapua.app.api.core.exception.model.SubjectUnauthorizedExceptionInfo;
 import org.eclipse.kapua.app.api.core.exception.model.ThrowableInfo;
+import org.eclipse.kapua.job.engine.JobStartOptions;
+import org.eclipse.kapua.job.engine.commons.model.JobStepPropertiesOverrides;
 import org.eclipse.kapua.job.engine.commons.model.JobTargetSublist;
 import org.eclipse.kapua.service.authentication.AuthenticationXmlRegistry;
 import org.eclipse.kapua.service.authentication.token.AccessToken;
@@ -161,7 +163,10 @@ public JaxbContextResolver() {
                     JobTargetQuery.class,
                     JobExecutionXmlRegistry.class,
 
+                    // Job Engine
+                    JobStartOptions.class,
                     JobTargetSublist.class,
+                    JobStepPropertiesOverrides.class,
 
                     DeviceCommandInput.class,
                     DevicePackageDownloadRequest.class,

diff --git a/...ngine/client/src/main/java/org/eclipse/kapua/job/engine/client/JobStartOptionsClient.java b/...ngine/client/src/main/java/org/eclipse/kapua/job/engine/client/JobStartOptionsClient.java
@@ -14,8 +14,11 @@
 
 import org.eclipse.kapua.job.engine.JobStartOptions;
 import org.eclipse.kapua.model.id.KapuaId;
+import org.eclipse.kapua.service.job.step.definition.JobStepProperty;
 
+import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -28,6 +31,7 @@ public class JobStartOptionsClient implements JobStartOptions {
     private static final long serialVersionUID = 7890046123237425642L;
 
     private Set<KapuaId> targetIdSublist;
+    private List<JobStepProperty> stepPropertiesOverrides;
     private boolean resetStepIndex;
     private Integer fromStepIndex;
     private boolean enqueue;
@@ -56,6 +60,25 @@ public void removeTargetIdToSublist(KapuaId targetId) {
         getTargetIdSublist().remove(targetId);
     }
 
+    @Override
+    public List<JobStepProperty> getStepPropertiesOverrides() {
+        if (stepPropertiesOverrides == null) {
+            stepPropertiesOverrides = new ArrayList<>();
+        }
+
+        return stepPropertiesOverrides;
+    }
+
+    @Override
+    public void addStepPropertyOverride(JobStepProperty jobStepPropertyOverride) {
+        getStepPropertiesOverrides().add(jobStepPropertyOverride);
+    }
+
+    @Override
+    public void setStepPropertiesOverrides(List<JobStepProperty> stepPropertiesOverrides) {
+        this.stepPropertiesOverrides = stepPropertiesOverrides;
+    }
+
     @Override
     public boolean getResetStepIndex() {
         return resetStepIndex;

diff --git a/.../src/main/java/org/eclipse/kapua/job/engine/commons/model/JobStepPropertiesOverrides.java b/.../src/main/java/org/eclipse/kapua/job/engine/commons/model/JobStepPropertiesOverrides.java
@@ -0,0 +1,65 @@
+/*******************************************************************************
+ * Copyright (c) 2018, 2022 Eurotech and/or its affiliates and others
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *
+ * Contributors:
+ *     Eurotech - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.kapua.job.engine.commons.model;
+
+import org.eclipse.kapua.service.job.step.definition.JobStepProperty;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlElementWrapper;
+import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.XmlType;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+@XmlRootElement(name = "jobStepPropertiesOverrides")
+@XmlAccessorType(XmlAccessType.FIELD)
+@XmlType
+public class JobStepPropertiesOverrides implements Iterable<JobStepProperty> {
+
+    @XmlElementWrapper(name = "jobStepPropertiesOverrides")
+    private List<JobStepProperty> jobStepPropertiesOverrides = new ArrayList<>();
+
+    public JobStepPropertiesOverrides() {
+    }
+
+    public JobStepPropertiesOverrides(List<JobStepProperty> targetIds) {
+        this.jobStepPropertiesOverrides.addAll(targetIds);
+    }
+
+    public List<JobStepProperty> getJobStepPropertiesOverrides() {
+        if (jobStepPropertiesOverrides == null) {
+            jobStepPropertiesOverrides = new ArrayList<>();
+        }
+
+        return jobStepPropertiesOverrides;
+    }
+
+    public void setJobStepPropertiesOverrides(List<JobStepProperty> jobStepPropertiesOverrides) {
+        this.jobStepPropertiesOverrides = jobStepPropertiesOverrides;
+    }
+
+    @Override
+    public Iterator<JobStepProperty> iterator() {
+        return jobStepPropertiesOverrides.iterator();
+    }
+
+    public boolean isEmpty() {
+        return getJobStepPropertiesOverrides().isEmpty();
+    }
+
+    public int size() {
+        return getJobStepPropertiesOverrides().size();
+    }
+}
diff --git a/.../src/main/java/org/eclipse/kapua/job/engine/commons/wrappers/JobContextPropertyNames.java b/.../src/main/java/org/eclipse/kapua/job/engine/commons/wrappers/JobContextPropertyNames.java
@@ -42,6 +42,11 @@ private JobContextPropertyNames() {
      */
     public static final String RESUMED_KAPUA_EXECUTION_ID = "job.execution.resumedId";
 
+    /**
+     * @since 2.0.0
+     */
+    public static final String JOB_STEP_PROPERTIES_OVERRIDES = "job.step.properties.overrides";
+
     /**
      * since 1.1.0
      */

diff --git a/...ommons/src/main/java/org/eclipse/kapua/job/engine/commons/wrappers/JobContextWrapper.java b/...ommons/src/main/java/org/eclipse/kapua/job/engine/commons/wrappers/JobContextWrapper.java
@@ -17,6 +17,7 @@
 import org.eclipse.kapua.commons.util.xml.XmlUtil;
 import org.eclipse.kapua.job.engine.commons.exception.ReadJobPropertyException;
 import org.eclipse.kapua.job.engine.commons.logger.JobLogger;
+import org.eclipse.kapua.job.engine.commons.model.JobStepPropertiesOverrides;
 import org.eclipse.kapua.job.engine.commons.model.JobTargetSublist;
 import org.eclipse.kapua.job.engine.commons.model.JobTransientUserData;
 import org.eclipse.kapua.model.id.KapuaId;
@@ -91,6 +92,22 @@ public JobTargetSublist getTargetSublist() {
         }
     }
 
+    /**
+     * Gets the {@link JobStepPropertiesOverrides} of the {@link org.eclipse.kapua.service.job.execution.JobExecution}.
+     *
+     * @return The current {@link JobStepPropertiesOverrides} of the {@link org.eclipse.kapua.service.job.execution.JobExecution}.
+     * @since 2.0.0
+     */
+    public JobStepPropertiesOverrides getJobStepPropertiesOverrides() {
+        String jobStepPropertiesOverrides = getProperties().getProperty(JobContextPropertyNames.JOB_STEP_PROPERTIES_OVERRIDES);
+
+        try {
+            return XmlUtil.unmarshal(jobStepPropertiesOverrides, JobStepPropertiesOverrides.class);
+        } catch (JAXBException | SAXException e) {
+            throw new ReadJobPropertyException(e, JobContextPropertyNames.JOB_STEP_PROPERTIES_OVERRIDES, jobStepPropertiesOverrides);
+        }
+    }
+
     /**
      * Gets the {@link JobExecution} {@link KapuaId} to resume
      *

diff --git a/job-engine/jbatch/src/main/java/org/eclipse/kapua/job/engine/jbatch/JobStartOptionsImpl.java b/job-engine/jbatch/src/main/java/org/eclipse/kapua/job/engine/jbatch/JobStartOptionsImpl.java
@@ -14,8 +14,11 @@
 
 import org.eclipse.kapua.job.engine.JobStartOptions;
 import org.eclipse.kapua.model.id.KapuaId;
+import org.eclipse.kapua.service.job.step.definition.JobStepProperty;
 
+import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -26,6 +29,7 @@
 public class JobStartOptionsImpl implements JobStartOptions {
 
     private Set<KapuaId> targetIdSublist;
+    private List<JobStepProperty> stepPropertiesOverrides;
     private boolean resetStepIndex;
     private Integer fromStepIndex;
     private boolean enqueue;
@@ -77,6 +81,25 @@ public void removeTargetIdToSublist(KapuaId targetId) {
         getTargetIdSublist().remove(targetId);
     }
 
+    @Override
+    public List<JobStepProperty> getStepPropertiesOverrides() {
+        if (stepPropertiesOverrides == null) {
+            stepPropertiesOverrides = new ArrayList<>();
+        }
+
+        return stepPropertiesOverrides;
+    }
+
+    @Override
+    public void addStepPropertyOverride(JobStepProperty stepPropertyOverride) {
+        getStepPropertiesOverrides().add(stepPropertyOverride);
+    }
+
+    @Override
+    public void setStepPropertiesOverrides(List<JobStepProperty> stepPropertiesOverrides) {
+        this.stepPropertiesOverrides = stepPropertiesOverrides;
+    }
+
     @Override
     public boolean getResetStepIndex() {
         return resetStepIndex;

diff --git a/job-engine/jbatch/src/main/java/org/eclipse/kapua/job/engine/jbatch/driver/JbatchDriver.java b/job-engine/jbatch/src/main/java/org/eclipse/kapua/job/engine/jbatch/driver/JbatchDriver.java
@@ -164,7 +164,7 @@ public static void startJob(@NotNull KapuaId scopeId, @NotNull KapuaId jobId, @N
                     jslStep.setNextFromAttribute("step-" + (jobStep.getStepIndex() + 1));
                 }
 
-                jslStep.setProperties(JobDefinitionBuildUtils.buildStepProperties(jobStepDefinition, jobStep, jobStepIterator.hasNext()));
+                jslStep.setProperties(JobDefinitionBuildUtils.buildStepProperties(jobStepDefinition, jobStep, jobStepIterator.hasNext(), jobStartOptions.getStepPropertiesOverrides()));
 
                 jslExecutionElements.add(jslStep);
             }