Add shadow write support at BaseVersionedAspectResource and BaseAspec… (#535)

* Add shadow write support at BaseVersionedAspectResource and BaseAspectV2Resourc

* Default null

* Default null and add to delete

* Checkstyle

---------

Co-authored-by: Rakhi Agrawal <rakagrawal@linkedin.com>

Author: rakhiagr

restli-resources/src/main/java/com/linkedin/metadata/restli/BaseAspectV2Resource.java

@@ -39,7 +39,7 @@
 
 
 /**
- * This resource is intended to be used as top level resousce, instead of a sub resource.
+ * This resource is intended to be used as top level resource, instead of a sub resource.
  * For sub-resource please use {@link BaseVersionedAspectResource}
  *
  * @param <URN> must be a valid {@link Urn} type
@@ -78,6 +78,13 @@ protected BaseRestliAuditor getAuditor() {
   @Nonnull
   protected abstract BaseLocalDAO<ASPECT_UNION, URN> getLocalDAO();
 
+  /**
+   * Returns {@link BaseLocalDAO} for shadowing the aspect.
+   */
+  protected BaseLocalDAO<ASPECT_UNION, URN> getLocalShadowDAO() {
+    return null;
+  }
+
   /**
    * Get the ASPECT associated with URN.
    */
@@ -112,6 +119,10 @@ public Task<CreateResponse> create(@Nonnull URN urn, @Nonnull ASPECT aspect) {
     return RestliUtils.toTask(() -> {
       final AuditStamp auditStamp = getAuditor().requestAuditStamp(getContext().getRawRequestContext());
       getLocalDAO().add(urn, aspect, auditStamp);
+      BaseLocalDAO<ASPECT_UNION, URN> shadowLocalDao = getLocalShadowDAO();
+      if (shadowLocalDao != null) {
+        shadowLocalDao.add(urn, aspect, auditStamp);
+      }
       return new CreateResponse(HttpStatus.S_201_CREATED);
     });
   }
@@ -123,6 +134,10 @@ public Task<CreateResponse> createWithTracking(@Nonnull URN urn, @Nonnull ASPECT
     return RestliUtils.toTask(() -> {
       final AuditStamp auditStamp = getAuditor().requestAuditStamp(getContext().getRawRequestContext());
       getLocalDAO().add(urn, aspect, auditStamp, trackingContext, ingestionParams);
+      BaseLocalDAO<ASPECT_UNION, URN> shadowLocalDao = getLocalShadowDAO();
+      if (shadowLocalDao != null) {
+        shadowLocalDao.add(urn, aspect, auditStamp, trackingContext, ingestionParams);
+      }
       return new CreateResponse(HttpStatus.S_201_CREATED);
     });
   }
@@ -138,6 +153,10 @@ public Task<CreateKVResponse<URN, ASPECT>> createAndGet(@Nonnull URN urn,
     return RestliUtils.toTask(() -> {
       final AuditStamp auditStamp = getAuditor().requestAuditStamp(getContext().getRawRequestContext());
       final ASPECT newValue = getLocalDAO().add(urn, _aspectClass, createLambda, auditStamp);
+      BaseLocalDAO<ASPECT_UNION, URN> shadowLocalDao = getLocalShadowDAO();
+      if (shadowLocalDao != null) {
+        shadowLocalDao.add(urn, _aspectClass, createLambda, auditStamp);
+      }
       return new CreateKVResponse<>(urn, newValue);
     });
   }
@@ -153,6 +172,10 @@ public Task<UpdateResponse> delete(@Nonnull URN urn) {
     return RestliUtils.toTask(() -> {
       final AuditStamp auditStamp = getAuditor().requestAuditStamp(getContext().getRawRequestContext());
       getLocalDAO().delete(urn, this._aspectClass, auditStamp);
+      BaseLocalDAO<ASPECT_UNION, URN> shadowLocalDao = getLocalShadowDAO();
+      if (shadowLocalDao != null) {
+        shadowLocalDao.delete(urn, this._aspectClass, auditStamp);
+      }
       return new UpdateResponse(HttpStatus.S_200_OK);
     });
   }

restli-resources/src/main/java/com/linkedin/metadata/restli/BaseVersionedAspectResource.java

@@ -71,6 +71,13 @@ protected BaseRestliAuditor getAuditor() {
   @Nonnull
   protected abstract BaseLocalDAO<ASPECT_UNION, URN> getLocalDAO();
 
+  /**
+   * Returns {@link BaseLocalDAO} for shadowing the aspect.
+   */
+  protected BaseLocalDAO<ASPECT_UNION, URN> getLocalShadowDAO() {
+    return null;
+  }
+
   /**
    * Constructs an entity-specific {@link Urn} based on the entity's {@link PathKeys}.
    */
@@ -109,6 +116,10 @@ public Task<CreateResponse> create(@Nonnull ASPECT aspect) {
       final URN urn = getUrn(getContext().getPathKeys());
       final AuditStamp auditStamp = getAuditor().requestAuditStamp(getContext().getRawRequestContext());
       getLocalDAO().add(urn, aspect, auditStamp);
+      BaseLocalDAO<ASPECT_UNION, URN> shadowLocalDao = getLocalShadowDAO();
+      if (shadowLocalDao != null) {
+        shadowLocalDao.add(urn, aspect, auditStamp);
+      }
       return new CreateResponse(HttpStatus.S_201_CREATED);
     });
   }
@@ -125,6 +136,10 @@ public Task<UpdateResponse> delete() {
       final URN urn = getUrn(getContext().getPathKeys());
       final AuditStamp auditStamp = getAuditor().requestAuditStamp(getContext().getRawRequestContext());
       getLocalDAO().delete(urn, this._aspectClass, auditStamp);
+      BaseLocalDAO<ASPECT_UNION, URN> shadowLocalDao = getLocalShadowDAO();
+      if (shadowLocalDao != null) {
+        shadowLocalDao.delete(urn, this._aspectClass, auditStamp);
+      }
       return new UpdateResponse(HttpStatus.S_200_OK);
     });
   }
@@ -139,6 +154,10 @@ public Task<CreateResponse> create(@Nonnull Class<ASPECT> aspectClass,
       final URN urn = getUrn(getContext().getPathKeys());
       final AuditStamp auditStamp = getAuditor().requestAuditStamp(getContext().getRawRequestContext());
       getLocalDAO().add(urn, aspectClass, createLambda, auditStamp);
+      BaseLocalDAO<ASPECT_UNION, URN> shadowLocalDao = getLocalShadowDAO();
+      if (shadowLocalDao != null) {
+        shadowLocalDao.add(urn, aspectClass, createLambda, auditStamp);
+      }
       return new CreateResponse(HttpStatus.S_201_CREATED);
     });
   }
@@ -156,6 +175,10 @@ public Task<CreateKVResponse<Long, ASPECT>> createAndGet(@Nonnull Class<ASPECT>
       final URN urn = getUrn(getContext().getPathKeys());
       final AuditStamp auditStamp = getAuditor().requestAuditStamp(getContext().getRawRequestContext());
       final ASPECT newValue = getLocalDAO().add(urn, aspectClass, createLambda, auditStamp);
+      BaseLocalDAO<ASPECT_UNION, URN> shadowLocalDao = getLocalShadowDAO();
+      if (shadowLocalDao != null) {
+        shadowLocalDao.add(urn, aspectClass, createLambda, auditStamp);
+      }
       return new CreateKVResponse<>(LATEST_VERSION, newValue);
     });
   }