bugfix: fixt restart streaming job and list scheduled jobs

caraml-dev · May 17, 2024 · 655dad7 · 655dad7
1 parent b7163a3
commit 655dad7
Show file tree

Hide file tree

Showing 6 changed files with 305 additions and 288 deletions.
diff --git a/caraml-store-protobuf/src/main/proto/feast_spark/api/JobService.proto b/caraml-store-protobuf/src/main/proto/feast_spark/api/JobService.proto
@@ -108,6 +108,9 @@ message Job {
 
   // Spark job error message, if available
   string error_message = 10;
+
+  // Project
+  string project = 11;
 }
 
 // Ingest data from offline store into online store

diff --git a/caraml-store-registry/src/main/java/dev/caraml/store/sparkjob/JobService.java b/caraml-store-registry/src/main/java/dev/caraml/store/sparkjob/JobService.java
@@ -120,11 +120,13 @@ private Job sparkApplicationToJob(SparkApplication app) {
             default -> JobStatus.JOB_STATUS_PENDING;
           };
     }
+    String project = labels.getOrDefault(PROJECT_LABEL, "");
     String tableName = labels.getOrDefault(FEATURE_TABLE_LABEL, "");
 
     Job.Builder builder =
         Job.newBuilder()
             .setId(app.getMetadata().getName())
+            .setProject(project)
             .setStartTime(startTime)
             .setStatus(jobStatus);
     switch (JobType.valueOf(labels.get(JOB_TYPE_LABEL))) {
@@ -162,7 +164,7 @@ private ScheduledJob scheduledSparkApplicationToScheduledJob(ScheduledSparkAppli
   public Job createOrUpdateStreamingIngestionJob(String project, String featureTableName) {
     FeatureTableSpec spec =
         tableRepository
-            .findFeatureTableByNameAndProject_NameAndIsDeletedFalse(project, featureTableName)
+            .findFeatureTableByNameAndProject_NameAndIsDeletedFalse(featureTableName, project)
             .map(ft -> ft.toProto().getSpec())
             .orElseThrow(
                 () -> {