check store lastUpdate (#846)

core/src/main/java/feast/core/model/Store.java

@@ -47,7 +47,7 @@
 @AllArgsConstructor
 @Entity
 @Table(name = "stores")
-public class Store {
+public class Store extends AbstractTimestampEntity {
 
   // Name of the store. Must be unique
   @Id

core/src/main/java/feast/core/service/JobCoordinatorService.java

@@ -227,6 +227,10 @@ private boolean jobRequiresUpgrade(Job job, Set<Store> stores) {
       return true;
     }
 
+    if (stores.stream().anyMatch(s -> s.getLastUpdated().after(job.getCreated()))) {
+      return true;
+    }
+
     return false;
   }
 

core/src/main/resources/db/migration/V2.4__Store_Timestamps.sql

@@ -0,0 +1,2 @@
+ALTER TABLE stores ADD COLUMN created timestamp default now();
+ALTER TABLE stores ADD COLUMN last_updated timestamp default now();

core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java

@@ -55,6 +55,7 @@
 import feast.proto.core.SourceProto.SourceType;
 import feast.proto.core.StoreProto;
 import feast.proto.core.StoreProto.Store.Subscription;
+import java.time.Instant;
 import java.util.*;
 import java.util.concurrent.CancellationException;
 import lombok.SneakyThrows;
@@ -749,6 +750,35 @@ public void shouldUpgradeJobWhenNeeded() {
     assertThat("CreateTask is expected", tasks.get(0) instanceof CreateJobTask);
   }
 
+  @Test
+  public void shouldUpgradeJobWhenStoreChanged() {
+    Source source = TestUtil.createKafkaSource("kafka:9092", "topic", false);
+    Store store = TestUtil.createStore("store", Collections.emptyList());
+
+    Job job =
+        Job.builder()
+            .setStatus(JobStatus.RUNNING)
+            .setFeatureSetJobStatuses(new HashSet<>())
+            .setSource(source)
+            .setStores(ImmutableSet.of(store))
+            .setExtId("extId")
+            .build();
+
+    job.setCreated(Date.from(Instant.now()));
+    store.setLastUpdated(Date.from(Instant.now().plusSeconds(1)));
+
+    when(jobRepository
+            .findFirstBySourceTypeAndSourceConfigAndStoreNameAndStatusNotInOrderByLastUpdatedDesc(
+                source.getType(), source.getConfig(), null, JobStatus.getTerminalStates()))
+        .thenReturn(Optional.of(job));
+
+    List<JobTask> tasks =
+        jcsWithConsolidation.makeJobUpdateTasks(
+            ImmutableList.of(Pair.of(source, ImmutableSet.of(store))));
+
+    assertThat("CreateTask is expected", tasks.get(0) instanceof CreateJobTask);
+  }
+
   @Test
   public void shouldCreateJobIfNoRunning() {
     Source source = TestUtil.createKafkaSource("kafka:9092", "topic", false);