DS-1208: Support setName for user inputs and add JobInternalDelegateStep for delegating steps within same job

xyz-jobs/xyz-job-service/src/main/java/com/here/xyz/jobs/Job.java

@@ -29,13 +29,15 @@
 import static com.here.xyz.jobs.RuntimeInfo.State.RUNNING;
 import static com.here.xyz.jobs.RuntimeInfo.State.SUBMITTED;
 import static com.here.xyz.jobs.RuntimeInfo.State.SUCCEEDED;
+import static com.here.xyz.jobs.steps.Step.InputSet.DEFAULT_INPUT_SET_NAME;
 import static com.here.xyz.jobs.steps.inputs.Input.inputS3Prefix;
 import static com.here.xyz.jobs.steps.resources.Load.addLoads;
 import static com.here.xyz.util.Random.randomAlpha;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonView;
 import com.here.xyz.XyzSerializable;
 import com.here.xyz.jobs.RuntimeInfo.State;
@@ -71,6 +73,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -88,6 +91,8 @@ public class Job implements XyzSerializable {
   private long updatedAt;
   @JsonView({Public.class, Static.class})
   private long keepUntil;
+  @JsonProperty(access = JsonProperty.Access.WRITE_ONLY)
+  private Map<String, Input> inputs;
   //Caller defined properties:
   @JsonView(Static.class)
   private String owner;
@@ -212,7 +217,7 @@ private Future<Void> prepareStep(Step step) {
   protected Future<Boolean> validate() {
     logger.info("[{}] Validating job ...", getId());
     //TODO: Collect exceptions and forward them accordingly as one exception object with (potentially) multiple error objects inside
-    return Future.all(Job.forEach(getSteps().stepStream().toList(), step -> validateStep(step)))
+    return Future.all(Job.forEach(nonFinalSteps().toList(), step -> validateStep(step)))
         .compose(cf -> Future.succeededFuture(cf.list().stream().allMatch(isReady -> (boolean) isReady)));
   }
 
@@ -234,7 +239,7 @@ public Future<Void> start() {
       return Future.failedFuture(new IllegalStateException("Job can not be started as it's not in SUBMITTED state."));
 
     getStatus().setState(PENDING);
-    getSteps().stepStream().forEach(step -> step.getStatus().setState(PENDING));
+    nonFinalSteps().forEach(step -> step.getStatus().setState(PENDING));
 
     long t1 = Core.currentTimeMillis();
     return store()
@@ -286,12 +291,16 @@ public Step getStepById(String stepId) {
     return getSteps().getStep(stepId);
   }
 
+  private Stream<Step> nonFinalSteps() {
+    return getSteps().stepStream().filter(step -> !step.getStatus().getState().isFinal());
+  }
+
   /**
    * Updates the status of a step at this job by replacing it with the specified one.
    * @param step
    * @return
    */
-  public Future<Void> updateStep(Step<?> step) {
+  public Future<Void> updateStep(Step step) {
     final Step existingStep = getStepById(step.getId());
     if (existingStep == null)
       throw new IllegalArgumentException("The provided step with ID " + step.getGlobalStepId() + " was not found.");
@@ -490,9 +499,13 @@ private Future<Map<ExecutionResource, Double>> calculateResourceLoads(Step step)
   }
 
   public UploadUrl createUploadUrl(boolean compressed) {
+    return createUploadUrl(compressed, DEFAULT_INPUT_SET_NAME);
+  }
+
+  public UploadUrl createUploadUrl(boolean compressed, String setName) {
     return new UploadUrl()
         .withCompressed(compressed)
-        .withS3Key(inputS3Prefix(getId()) + "/" + UUID.randomUUID() + (compressed ? ".gz" : ""));
+        .withS3Key(inputS3Prefix(getId(), setName) + "/" + UUID.randomUUID() + (compressed ? ".gz" : ""));
   }
 
   public Future<Void> consumeInput(ModelBasedInput input) {
@@ -510,8 +523,8 @@ private Future<Void> deleteInputs() {
     return Future.succeededFuture();
   }
 
-  public Future<List<Input>> loadInputs() {
-    return ASYNC.run(() -> Input.loadInputs(getId()));
+  public Future<List<Input>> loadInputs(String setName) {
+    return ASYNC.run(() -> Input.loadInputs(getId(), setName));
   }
 
   public Future<List<Output>> loadOutputs() {
@@ -637,6 +650,19 @@ public Job withKeepUntil(long keepUntil) {
     return this;
   }
 
+  public Map<String, Input> getInputs() {
+    return inputs;
+  }
+
+  public void setInputs(Map<String, Input> inputs) {
+    this.inputs = inputs;
+  }
+
+  public Job withInputs(Map<String, Input> inputs) {
+    setInputs(inputs);
+    return this;
+  }
+
   public String getOwner() {
     return owner;
   }

xyz-jobs/xyz-job-service/src/main/java/com/here/xyz/jobs/service/JobAdminApi.java

@@ -32,6 +32,9 @@
 import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
 import static io.netty.handler.codec.http.HttpResponseStatus.NO_CONTENT;
 import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.vertx.core.http.HttpMethod.DELETE;
+import static io.vertx.core.http.HttpMethod.GET;
+import static io.vertx.core.http.HttpMethod.POST;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.here.xyz.XyzSerializable;
@@ -42,7 +45,6 @@
 import com.here.xyz.jobs.steps.execution.JobExecutor;
 import com.here.xyz.util.service.HttpException;
 import io.vertx.core.Future;
-import io.vertx.core.http.HttpMethod;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.Router;
 import io.vertx.ext.web.RoutingContext;
@@ -60,12 +62,12 @@ public class JobAdminApi extends JobApiBase {
   private static final String ADMIN_STATE_MACHINE_EVENTS = "/admin/state/events";
 
   public JobAdminApi(Router router) {
-    router.route(HttpMethod.GET, ADMIN_JOBS).handler(handleErrors(this::getJobs));
-    router.route(HttpMethod.GET, ADMIN_JOB).handler(handleErrors(this::getJob));
-    router.route(HttpMethod.DELETE, ADMIN_JOBS).handler(handleErrors(this::deleteJob));
-    router.route(HttpMethod.POST, ADMIN_JOB_STEPS).handler(handleErrors(this::postStep));
-    router.route(HttpMethod.GET, ADMIN_JOB_STEP).handler(handleErrors(this::getStep));
-    router.route(HttpMethod.POST, ADMIN_STATE_MACHINE_EVENTS).handler(handleErrors(this::postStateEvent));
+    router.route(GET, ADMIN_JOBS).handler(handleErrors(this::getJobs));
+    router.route(GET, ADMIN_JOB).handler(handleErrors(this::getJob));
+    router.route(DELETE, ADMIN_JOBS).handler(handleErrors(this::deleteJob));
+    router.route(POST, ADMIN_JOB_STEPS).handler(handleErrors(this::postStep));
+    router.route(GET, ADMIN_JOB_STEP).handler(handleErrors(this::getStep));
+    router.route(POST, ADMIN_STATE_MACHINE_EVENTS).handler(handleErrors(this::postStateEvent));
   }
 
   private void getJobs(RoutingContext context) {

xyz-jobs/xyz-job-service/src/main/java/com/here/xyz/jobs/service/JobApi.java

@@ -19,11 +19,13 @@
 
 package com.here.xyz.jobs.service;
 
+import static com.here.xyz.jobs.RuntimeInfo.State.FAILED;
 import static com.here.xyz.jobs.RuntimeInfo.State.NOT_READY;
 import static com.here.xyz.jobs.RuntimeInfo.State.RUNNING;
 import static com.here.xyz.jobs.RuntimeStatus.Action.CANCEL;
 import static com.here.xyz.jobs.service.JobApiBase.ApiParam.Path.SPACE_ID;
 import static com.here.xyz.jobs.service.JobApiBase.ApiParam.getPathParam;
+import static com.here.xyz.jobs.steps.Step.InputSet.DEFAULT_INPUT_SET_NAME;
 import static io.netty.handler.codec.http.HttpResponseStatus.ACCEPTED;
 import static io.netty.handler.codec.http.HttpResponseStatus.CREATED;
 import static io.netty.handler.codec.http.HttpResponseStatus.OK;
@@ -48,7 +50,6 @@
 import io.vertx.core.Future;
 import io.vertx.ext.web.RoutingContext;
 import io.vertx.ext.web.openapi.router.RouterBuilder;
-import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -67,6 +68,8 @@ public JobApi(RouterBuilder rb) {
     rb.getRoute("deleteJob").setDoValidation(false).addHandler(handleErrors(this::deleteJob));
     rb.getRoute("postJobInputs").setDoValidation(false).addHandler(handleErrors(this::postJobInput));
     rb.getRoute("getJobInputs").setDoValidation(false).addHandler(handleErrors(this::getJobInputs));
+    rb.getRoute("postNamedJobInputs").setDoValidation(false).addHandler(handleErrors(this::postJobInput));
+    rb.getRoute("getNamedJobInputs").setDoValidation(false).addHandler(handleErrors(this::getJobInputs));
     rb.getRoute("getJobOutputs").setDoValidation(false).addHandler(handleErrors(this::getJobOutputs));
     rb.getRoute("patchJobStatus").setDoValidation(false).addHandler(handleErrors(this::patchJobStatus));
     rb.getRoute("getJobStatus").setDoValidation(false).addHandler(handleErrors(this::getJobStatus));
@@ -79,6 +82,7 @@ protected void postJob(final RoutingContext context) throws HttpException {
   protected Future<Job> createNewJob(RoutingContext context, Job job) {
     logger.info(getMarker(context), "Received job creation request: {}", job.serialize(true));
     return job.create().submit()
+        .compose(v -> applyInputReferences(job))
         .map(res -> job)
         .recover(t -> {
           if (t instanceof CompilationError)
@@ -94,6 +98,34 @@ protected Future<Job> createNewJob(RoutingContext context, Job job) {
         .onFailure(err -> sendErrorResponse(context, err));
   }
 
+  protected Future<Void> applyInputReferences(Job job) {
+    if (job.getInputs() == null)
+      return Future.succeededFuture();
+
+    if (!job.getInputs().values().stream().allMatch(input -> input instanceof InputsFromS3))
+      return Future.failedFuture("Only inputs of type " + InputsFromS3.class.getSimpleName() + " are supported as inline inputs.");
+
+    //Continue with the input scanning *asynchronously* but fail the job if something goes wrong (User can check the status)
+    Future.all(job.getInputs().entrySet().stream()
+        .map(inputSet -> registerInput(job, (InputsFromS3) inputSet.getValue(), inputSet.getKey()))
+        .toList())
+        .onFailure(t -> {
+          logger.error("[{}] Error while scanning inputs for job.", job.getId(), t);
+          job.getStatus()
+              .withState(FAILED)
+              .withErrorMessage("Error while scanning inputs.")
+              .withErrorCause(t.getMessage());
+          job.store();
+        })
+        .compose(v -> job.submit());
+
+    /*
+    Return without waiting for the input scanning to complete.
+    The job will stay in state NOT_READY for some time but will proceed automatically afterwards.
+     */
+    return Future.succeededFuture();
+  }
+
   protected void getJobs(final RoutingContext context) {
     getJobs(context, false);
   }
@@ -116,67 +148,74 @@ protected void deleteJob(final RoutingContext context) {
   protected void postJobInput(final RoutingContext context) throws HttpException {
     String jobId = jobId(context);
     Input input = getJobInputFromBody(context);
-    if (input instanceof UploadUrl uploadUrl) {
-      loadJob(context, jobId)
-          .compose(job -> job.getStatus().getState() == NOT_READY
-              ? Future.succeededFuture(job)
-              : Future.failedFuture(new DetailedHttpException("E319004")))
-          .map(job -> job.createUploadUrl(uploadUrl.isCompressed()))
-          .onSuccess(res -> sendResponse(context, CREATED.code(), res))
-          .onFailure(err -> sendErrorResponse(context, err));
-    }
-    else if (input instanceof InputsFromS3 s3Inputs) {
-      loadJob(context, jobId)
-          .compose(job -> job.getStatus().getState() == NOT_READY
-              ? Future.succeededFuture(job)
-              : Future.failedFuture(new DetailedHttpException("E319004")))
-          .compose(job -> {
-            s3Inputs.dereference(job.getId());
-            return Future.succeededFuture();
-          })
-          .onSuccess(v -> sendResponse(context, OK.code(), (XyzSerializable) null))
-          .onFailure(err -> sendErrorResponse(context, err));
-    }
-    else if (input instanceof ModelBasedInput modelBasedInput) {
-      loadJob(context, jobId)
-          .compose(job -> {
-            if (!job.isPipeline())
-              return Future.failedFuture(new DetailedHttpException("E319005", Map.of("allowedType", UploadUrl.class.getSimpleName())));
-            else if (job.getStatus().getState() != RUNNING)
-              return Future.failedFuture(new DetailedHttpException("E319006"));
-            else if (context.request().bytesRead() > 256 * 1024)
-              return Future.failedFuture(new DetailedHttpException("E319007"));
-            else
-              return job.consumeInput(modelBasedInput);
-          })
-          .onSuccess(v -> sendResponse(context, OK.code(), (XyzSerializable) null))
-          .onFailure(err -> sendErrorResponse(context, err));
-    }
-    else if (input instanceof InputsFromJob inputsReference) {
-      //NOTE: Both jobs have to be loaded to authorize the user for both
-      loadJob(context, jobId)
-          .compose(job -> loadJob(context, inputsReference.getJobId()).compose(referencedJob -> {
-            try {
-              if (!Objects.equals(referencedJob.getOwner(), job.getOwner()))
-                return Future.failedFuture(new DetailedHttpException("E319008", Map.of("referencedJob", inputsReference.getJobId(), "referencingJob", job.getId())));
-
-              inputsReference.dereference(job.getId());
-              return Future.succeededFuture();
-            }
-            catch (IOException e) {
-              return Future.failedFuture(e);
-            }
-          }))
-          .onSuccess(v -> sendResponse(context, OK.code(), (XyzSerializable) null))
-          .onFailure(err -> sendErrorResponse(context, err));
-    }
+    String inputSetName = inputSetName(context);
+
+    Future<Input> inputCreatedFuture = loadJob(context, jobId).compose(job -> registerInput(context, job, input, inputSetName));
+
+    inputCreatedFuture
+        .onSuccess(res -> sendResponse(context, CREATED.code(), res))
+        .onFailure(err -> sendErrorResponse(context, err));
+  }
+
+  private Future<Input> registerInput(RoutingContext context, Job job, Input input, String inputSetName) {
+    if (input instanceof UploadUrl uploadUrl)
+      return registerInput(job, inputSetName, uploadUrl);
+
+    if (input instanceof InputsFromS3 s3Inputs)
+      return registerInput(job, s3Inputs, inputSetName);
+
+    if (input instanceof ModelBasedInput modelBasedInput)
+      return registerPipelineInput(context, job, modelBasedInput);
+
+    if (input instanceof InputsFromJob inputsReference)
+      return registerInput(context, job, inputsReference);
+
+    throw new NotImplementedException("Input type " + input.getClass().getSimpleName() + " is not supported.");
+  }
+
+  private static Future<Input> registerInput(Job job, String inputSetName, UploadUrl uploadUrl) {
+    return job.getStatus().getState() == NOT_READY
+        ? Future.succeededFuture(job.createUploadUrl(uploadUrl.isCompressed(), inputSetName))
+        : Future.failedFuture(new DetailedHttpException("E319004"));
+  }
+
+  private Future<Input> registerInput(RoutingContext context, Job job, InputsFromJob inputsReference) {
+    //NOTE: Both jobs have to be loaded to authorize the user for both ones
+    return loadJob(context, inputsReference.getJobId()).compose(referencedJob -> {
+      try {
+        if (!Objects.equals(referencedJob.getOwner(), job.getOwner()))
+          return Future.failedFuture(new DetailedHttpException("E319008", Map.of("referencedJob", inputsReference.getJobId(), "referencingJob", job.getId())));
+
+        inputsReference.dereference(job.getId());
+        return Future.succeededFuture();
+      }
+      catch (Exception e) {
+        return Future.failedFuture(e);
+      }
+    });
+  }
+
+  private static Future<Input> registerPipelineInput(RoutingContext context, Job job, ModelBasedInput modelBasedInput) {
+    if (!job.isPipeline())
+      return Future.failedFuture(new DetailedHttpException("E319005", Map.of("allowedType", UploadUrl.class.getSimpleName())));
+    else if (job.getStatus().getState() != RUNNING)
+      return Future.failedFuture(new DetailedHttpException("E319006"));
+    else if (context.request().bytesRead() > 256 * 1024)
+      return Future.failedFuture(new DetailedHttpException("E319007"));
     else
-      throw new NotImplementedException("Input type " + input.getClass().getSimpleName() + " is not supported.");
+      return job.consumeInput(modelBasedInput).map(null);
+  }
+
+  private static Future<Input> registerInput(Job job, InputsFromS3 s3Inputs, String inputSetName) {
+    if (job.getStatus().getState() != NOT_READY)
+      return Future.failedFuture(new DetailedHttpException("E319004"));
+    s3Inputs.dereference(job.getId(), inputSetName);
+    return Future.succeededFuture(null);
   }
 
   protected void getJobInputs(final RoutingContext context) {
     loadJob(context, jobId(context))
-        .compose(job -> job.loadInputs())
+        .compose(job -> job.loadInputs(inputSetName(context)))
         .onSuccess(res -> sendResponse(context, OK.code(), res, new TypeReference<List<Input>>() {}))
         .onFailure(err -> sendErrorResponse(context, err));
   }
@@ -234,6 +273,11 @@ protected Future<Void> authorizeAccess(RoutingContext context, Job job) {
     return Future.succeededFuture();
   }
 
+  protected String inputSetName(RoutingContext context) {
+    String setName = context.pathParam("setName");
+    return setName == null ? DEFAULT_INPUT_SET_NAME : setName;
+  }
+
   protected Input getJobInputFromBody(RoutingContext context) throws HttpException {
     try {
       try {

xyz-jobs/xyz-job-service/src/main/java/com/here/xyz/jobs/steps/StepGraph.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2017-2024 HERE Europe B.V.
+ * Copyright (C) 2017-2025 HERE Europe B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.