Repository: gojek/feast (java)

Issue:
Feast Serving not registering store in Feast Core
## Expected Behavior 
Feast Serving should register its store in Feast Core once it starts up.
```
grpc_cli call localhost:6565 ListStores ''
```
`ListStores` on CoreService should show the store registered by Feast Serving:
```
store {
  name: "online"
  type: REDIS
  // ......
}
```

## Current Behavior
Feast Serving does not register its store in Feast core and the store does not show 
up in Feast Core when calling `ListStores` on CoreService.

Pull request description:
Fix Feast Serving not registering its store in Feast Core
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. Ensure that your code follows our code conventions: https://github.com/gojek/feast/blob/master/docs/contributing.md#code-conventions
2. Run unit tests and ensure that they are passing: https://github.com/gojek/feast/blob/master/docs/contributing.md#running-unit-tests
3. If your change introduces any API changes, make sure to update the integration tests scripts here: https://github.com/gojek/feast/tree/master/tests/e2e
4. Make sure documentation is updated for your PR!
5. Make sure you have signed the CLA https://cla.developers.google.com/clas

-->

**What this PR does / why we need it**:
Fixes Feast Serving not registering its store into Feast Core.
- Adds `registerSource()` to `CoreSpecService` to register source spec in in Feast Core.
- Adds missing `registerSource()` call in `CachedSpecService` to register its store in Feast Core.
- Adds a unit test to ensure that cached spec service registers its store with Feast Core.

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes #640 

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".

For more information about release notes, see kubernetes' guide here:
http://git.k8s.io/community/contributors/guide/release-notes.md
-->
```release-note
NONE
```


Patch:
diff --git a/serving/src/main/java/feast/serving/specs/CachedSpecService.java b/serving/src/main/java/feast/serving/specs/CachedSpecService.java
index 2f68711bf..a117754e8 100644
--- a/serving/src/main/java/feast/serving/specs/CachedSpecService.java
+++ b/serving/src/main/java/feast/serving/specs/CachedSpecService.java
@@ -47,7 +47,7 @@ import java.util.stream.Collectors;
 import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 
-/** In-memory cache of specs. */
+/** In-memory cache of specs hosted in Feast Core. */
 public class CachedSpecService {
 
   private static final int MAX_SPEC_COUNT = 1000;
@@ -76,7 +76,7 @@ public class CachedSpecService {
 
   public CachedSpecService(CoreSpecService coreService, StoreProto.Store store) {
     this.coreService = coreService;
-    this.store = store;
+    this.store = coreService.registerStore(store);
 
     Map<String, FeatureSetSpec> featureSets = getFeatureSetMap();
     featureToFeatureSetMapping =
diff --git a/serving/src/main/java/feast/serving/specs/CoreSpecService.java b/serving/src/main/java/feast/serving/specs/CoreSpecService.java
index 2f5cef342..259aa3f3f 100644
--- a/serving/src/main/java/feast/serving/specs/CoreSpecService.java
+++ b/serving/src/main/java/feast/serving/specs/CoreSpecService.java
@@ -23,11 +23,12 @@ import feast.core.CoreServiceProto.ListFeatureSetsRequest;
 import feast.core.CoreServiceProto.ListFeatureSetsResponse;
 import feast.core.CoreServiceProto.UpdateStoreRequest;
 import feast.core.CoreServiceProto.UpdateStoreResponse;
+import feast.core.StoreProto.Store;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import org.slf4j.Logger;
 
-/** Client for spec retrieval from core. */
+/** Client for interfacing with specs in Feast Core. */
 public class CoreSpecService {
 
   private static final Logger log = org.slf4j.LoggerFactory.getLogger(CoreSpecService.class);
@@ -50,4 +51,24 @@ public class CoreSpecService {
   public UpdateStoreResponse updateStore(UpdateStoreRequest updateStoreRequest) {
     return blockingStub.updateStore(updateStoreRequest);
   }
+
+  /**
+   * Register the given store entry in Feast Core. If store already exists in Feast Core, updates
+   * the store entry in feast core.
+   *
+   * @param store entry to register/update in Feast Core.
+   * @return The register/updated store entry
+   */
+  public Store registerStore(Store store) {
+    UpdateStoreRequest request = UpdateStoreRequest.newBuilder().setStore(store).build();
+    try {
+      UpdateStoreResponse updateStoreResponse = this.updateStore(request);
+      if (!updateStoreResponse.getStore().equals(store)) {
+        throw new RuntimeException("Core store config not matching current store config");
+      }
+      return updateStoreResponse.getStore();
+    } catch (Exception e) {
+      throw new RuntimeException("Unable to update store configuration", e);
+    }
+  }
 }


Test patch:
diff --git a/serving/src/test/java/feast/serving/service/CachedSpecServiceTest.java b/serving/src/test/java/feast/serving/service/CachedSpecServiceTest.java
index f4f795ed3..144b967c9 100644
--- a/serving/src/test/java/feast/serving/service/CachedSpecServiceTest.java
+++ b/serving/src/test/java/feast/serving/service/CachedSpecServiceTest.java
@@ -19,14 +19,14 @@ package feast.serving.service;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.mockito.MockitoAnnotations.initMocks;
 
 import com.google.common.collect.Lists;
 import feast.core.CoreServiceProto.ListFeatureSetsRequest;
 import feast.core.CoreServiceProto.ListFeatureSetsResponse;
-import feast.core.CoreServiceProto.UpdateStoreRequest;
-import feast.core.CoreServiceProto.UpdateStoreResponse;
 import feast.core.FeatureSetProto;
 import feast.core.FeatureSetProto.FeatureSetSpec;
 import feast.core.FeatureSetProto.FeatureSpec;
@@ -82,8 +82,7 @@ public class CachedSpecServiceTest {
                     .build())
             .build();
 
-    when(coreService.updateStore(UpdateStoreRequest.newBuilder().setStore(store).build()))
-        .thenReturn(UpdateStoreResponse.newBuilder().setStore(store).build());
+    when(coreService.registerStore(store)).thenReturn(store);
 
     featureSetSpecs = new LinkedHashMap<>();
     featureSetSpecs.put(
@@ -143,6 +142,11 @@ public class CachedSpecServiceTest {
     cachedSpecService = new CachedSpecService(coreService, store);
   }
 
+  @Test
+  public void shouldRegisterStoreWithCore() {
+    verify(coreService, times(1)).registerStore(cachedSpecService.getStore());
+  }
+
   @Test
   public void shouldPopulateAndReturnStore() {
     cachedSpecService.populateCache();