feast-dev
diff --git a/‎job-controller/src/main/java/feast/jobcontroller/config/FeastProperties.java‎
Lines changed: 4 additions & 0 deletions b/‎job-controller/src/main/java/feast/jobcontroller/config/FeastProperties.java‎
Lines changed: 4 additions & 0 deletions
diff --git a/‎job-controller/src/main/java/feast/jobcontroller/config/JobControllerConfig.java‎
Lines changed: 2 additions & 2 deletions b/‎job-controller/src/main/java/feast/jobcontroller/config/JobControllerConfig.java‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎job-controller/src/main/java/feast/jobcontroller/runner/ConsolidatedJobStrategy.java‎
Lines changed: 7 additions & 1 deletion b/‎job-controller/src/main/java/feast/jobcontroller/runner/ConsolidatedJobStrategy.java‎
Lines changed: 7 additions & 1 deletion
diff --git a/‎job-controller/src/main/java/feast/jobcontroller/runner/JobPerStoreStrategy.java‎
Lines changed: 7 additions & 1 deletion b/‎job-controller/src/main/java/feast/jobcontroller/runner/JobPerStoreStrategy.java‎
Lines changed: 7 additions & 1 deletion
diff --git a/‎job-controller/src/main/resources/application.yml‎
Lines changed: 3 additions & 0 deletions b/‎job-controller/src/main/resources/application.yml‎
Lines changed: 3 additions & 0 deletions
@@ -176,6 +176,10 @@ public feast.jobcontroller.runner.Runner getType() {
     /* Population job metric properties */
     private MetricsProperties metrics;
 
+    @NotNull
+    /* Prefix for JobId to separate consumer groups for independent jobs running in parallel */
+    private String jobIdPrefix;
+
     /* Timeout in seconds for each attempt to update or submit a new job to the runner */
     @Positive private long jobUpdateTimeoutSeconds;
 
 
@@ -87,9 +87,9 @@ public JobGroupingStrategy getJobGroupingStrategy(
     Boolean shouldConsolidateJobs =
         feastProperties.getJobs().getController().getConsolidateJobsPerSource();
     if (shouldConsolidateJobs) {
-      return new ConsolidatedJobStrategy(jobRepository);
+      return new ConsolidatedJobStrategy(jobRepository, feastProperties.getJobs());
     } else {
-      return new JobPerStoreStrategy(jobRepository);
+      return new JobPerStoreStrategy(jobRepository, feastProperties.getJobs());
     }
   }
 
 
@@ -16,6 +16,7 @@
  */
 package feast.jobcontroller.runner;
 
+import feast.jobcontroller.config.FeastProperties.JobProperties;
 import feast.jobcontroller.dao.JobRepository;
 import feast.jobcontroller.model.Job;
 import feast.jobcontroller.model.JobStatus;
@@ -38,9 +39,11 @@
  */
 public class ConsolidatedJobStrategy implements JobGroupingStrategy {
   private final JobRepository jobRepository;
+  private final JobProperties jobProperties;
 
-  public ConsolidatedJobStrategy(JobRepository jobRepository) {
+  public ConsolidatedJobStrategy(JobRepository jobRepository, JobProperties jobProperties) {
     this.jobRepository = jobRepository;
+    this.jobProperties = jobProperties;
   }
 
   @Override
@@ -71,6 +74,9 @@ private String createJobId(SourceProto.Source source) {
                 source.getKafkaSourceConfig().getBootstrapServers(),
                 source.getKafkaSourceConfig().getTopic()),
             dateSuffix);
+    if (!this.jobProperties.getJobIdPrefix().isEmpty()) {
+      jobId = this.jobProperties.getJobIdPrefix() + "-" + jobId;
+    }
     return jobId.replaceAll("_store", "-").toLowerCase();
   }
 
 
@@ -17,6 +17,7 @@
 package feast.jobcontroller.runner;
 
 import com.google.common.collect.Lists;
+import feast.jobcontroller.config.FeastProperties.JobProperties;
 import feast.jobcontroller.dao.JobRepository;
 import feast.jobcontroller.model.Job;
 import feast.jobcontroller.model.JobStatus;
@@ -38,9 +39,11 @@
  */
 public class JobPerStoreStrategy implements JobGroupingStrategy {
   private final JobRepository jobRepository;
+  private final JobProperties jobProperties;
 
-  public JobPerStoreStrategy(JobRepository jobRepository) {
+  public JobPerStoreStrategy(JobRepository jobRepository, JobProperties jobProperties) {
     this.jobRepository = jobRepository;
+    this.jobProperties = jobProperties;
   }
 
   @Override
@@ -78,6 +81,9 @@ private String createJobId(SourceProto.Source source, Iterable<StoreProto.Store>
                 source.getKafkaSourceConfig().getTopic()),
             Lists.newArrayList(stores).get(0).getName(),
             dateSuffix);
+    if (!this.jobProperties.getJobIdPrefix().isEmpty()) {
+      jobId = this.jobProperties.getJobIdPrefix() + "-" + jobId;
+    }
     return jobId.replaceAll("_store", "-").toLowerCase();
   }
 
 
@@ -23,6 +23,9 @@ feast:
     # Enabling JobManagement
     enabled: true
 
+    # Prefix for JobId
+    job_id_prefix: ""
+
     # Job update polling interval in milliseconds: how often Feast checks if new jobs should be sent to the runner.
     polling_interval_milliseconds: 60000
Original file line number	Diff line number	Diff line change
`@@ -87,9 +87,9 @@ public JobGroupingStrategy getJobGroupingStrategy(`
`87`	`87`	`Boolean shouldConsolidateJobs =`
`88`	`88`	`feastProperties.getJobs().getController().getConsolidateJobsPerSource();`
`89`	`89`	`if (shouldConsolidateJobs) {`
`90`		`- return new ConsolidatedJobStrategy(jobRepository);`
	`90`	`+ return new ConsolidatedJobStrategy(jobRepository, feastProperties.getJobs());`
`91`	`91`	`} else {`
`92`		`- return new JobPerStoreStrategy(jobRepository);`
	`92`	`+ return new JobPerStoreStrategy(jobRepository, feastProperties.getJobs());`
`93`	`93`	`}`
`94`	`94`	`}`
`95`	`95`