-
Notifications
You must be signed in to change notification settings - Fork 58
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactored RPC Routing for grpc implementation (#447)
* Refactored * Fixed the refactoring issues in the code * Fixed import * Added unit tests * Addressed comments * Addressed comments * Cleaned code * Changed to constructor --------- Co-authored-by: Rakhi Agrawal <[email protected]>
- Loading branch information
Showing
16 changed files
with
245 additions
and
233 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
40 changes: 0 additions & 40 deletions
40
.../main/java/com/linkedin/metadata/dao/ingestion/RestliCompliantPreUpdateRoutingClient.java
This file was deleted.
Oops, something went wrong.
24 changes: 0 additions & 24 deletions
24
dao-api/src/main/java/com/linkedin/metadata/dao/ingestion/RestliPreUpdateAspectRegistry.java
This file was deleted.
Oops, something went wrong.
44 changes: 44 additions & 0 deletions
44
.../src/main/java/com/linkedin/metadata/dao/ingestion/preupdate/PreUpdateAspectRegistry.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| package com.linkedin.metadata.dao.ingestion.preupdate; | ||
|
|
||
| import com.linkedin.data.template.RecordTemplate; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import javax.annotation.Nonnull; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
|
|
||
| /** | ||
| * A registry which maintains mapping of aspects and their PreUpdateRoutingClient. | ||
| */ | ||
| @Slf4j | ||
| public class PreUpdateAspectRegistry { | ||
|
|
||
| private final Map<Class<? extends RecordTemplate>, PreUpdateRoutingAccessor> _preUpdateLambdaMap; | ||
|
|
||
| /** | ||
| * Constructor to register pre-update routing accessors for multiple aspects at once. | ||
| * @param preUpdateMap map containing aspect classes and their corresponding accessors | ||
| */ | ||
| public PreUpdateAspectRegistry(@Nonnull Map<Class<? extends RecordTemplate>, PreUpdateRoutingAccessor> preUpdateMap) { | ||
| _preUpdateLambdaMap = new HashMap<>(preUpdateMap); | ||
| log.info("Registered pre-update routing accessors for aspects: {}", _preUpdateLambdaMap.keySet()); | ||
| } | ||
|
|
||
| /** | ||
| * Get Pre Update Routing Accessor for an aspect class. | ||
| * @param aspectClass the class of the aspect to retrieve the accessor for | ||
| * @return PreUpdateRoutingAccessor for the given aspect class, or null if not found | ||
| */ | ||
| public <ASPECT extends RecordTemplate> PreUpdateRoutingAccessor getPreUpdateRoutingAccessor( | ||
| @Nonnull Class<ASPECT> aspectClass) { | ||
| return _preUpdateLambdaMap.get(aspectClass); | ||
| } | ||
|
|
||
| /** | ||
| * Check if Pre Update Routing Accessor is registered for an aspect. | ||
| */ | ||
| public <ASPECT extends RecordTemplate> boolean isRegistered(@Nonnull final Class<ASPECT> aspectClass) { | ||
| return _preUpdateLambdaMap.containsKey(aspectClass); | ||
| } | ||
|
|
||
| } |
12 changes: 12 additions & 0 deletions
12
dao-api/src/main/java/com/linkedin/metadata/dao/ingestion/preupdate/PreUpdateResponse.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| package com.linkedin.metadata.dao.ingestion.preupdate; | ||
|
|
||
| import com.linkedin.data.template.RecordTemplate; | ||
| import lombok.Data; | ||
|
|
||
| /** | ||
| * Response of pre-update process that includes the updated aspect. | ||
| */ | ||
| @Data | ||
| public class PreUpdateResponse<ASPECT extends RecordTemplate> { | ||
| private final ASPECT updatedAspect; | ||
| } |
15 changes: 15 additions & 0 deletions
15
...src/main/java/com/linkedin/metadata/dao/ingestion/preupdate/PreUpdateRoutingAccessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| package com.linkedin.metadata.dao.ingestion.preupdate; | ||
|
|
||
| import com.linkedin.data.template.RecordTemplate; | ||
| import lombok.Data; | ||
|
|
||
|
|
||
| @Data | ||
| public class PreUpdateRoutingAccessor { | ||
|
|
||
| public PreUpdateRoutingClient<? extends RecordTemplate> preUpdateClient; | ||
|
|
||
| public enum RoutingAction { | ||
| PROCEED, SKIP | ||
| } | ||
| } |
9 changes: 5 additions & 4 deletions
9
...dao/ingestion/PreUpdateRoutingClient.java → ...ion/preupdate/PreUpdateRoutingClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,18 +1,19 @@ | ||
| package com.linkedin.metadata.dao.ingestion; | ||
| package com.linkedin.metadata.dao.ingestion.preupdate; | ||
|
|
||
| import com.google.protobuf.Message; | ||
| import com.linkedin.common.urn.Urn; | ||
| import com.linkedin.data.template.RecordTemplate; | ||
|
|
||
|
|
||
| /** | ||
| * An interface that defines methods to route update requests to the appropriate custom APIs for pre-ingestion process. | ||
| */ | ||
|
|
||
| public interface PreUpdateRoutingClient<ASPECT extends Message> { | ||
| public interface PreUpdateRoutingClient<ASPECT extends RecordTemplate> { | ||
| /** | ||
| * A method that routes the update request to the appropriate custom API. | ||
| * @param urn the urn of the asset | ||
| * @param aspect the aspect to be updated | ||
| * @return the updated aspect | ||
| */ | ||
| ASPECT routingLambda(Message urn, ASPECT aspect); | ||
| PreUpdateResponse<ASPECT> preUpdate(Urn urn, ASPECT aspect); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
28 changes: 0 additions & 28 deletions
28
.../src/test/java/com/linkedin/metadata/dao/ingestion/SamplePreUpdateAspectRegistryImpl.java
This file was deleted.
Oops, something went wrong.
36 changes: 7 additions & 29 deletions
36
dao-api/src/test/java/com/linkedin/metadata/dao/ingestion/SamplePreUpdateRoutingClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,40 +1,18 @@ | ||
| package com.linkedin.metadata.dao.ingestion; | ||
|
|
||
| import com.google.protobuf.Any; | ||
| import com.google.protobuf.Message; | ||
| import com.google.protobuf.StringValue; | ||
| import com.linkedin.common.urn.Urn; | ||
| import com.linkedin.data.template.RecordTemplate; | ||
| import com.linkedin.metadata.dao.ingestion.preupdate.PreUpdateResponse; | ||
| import com.linkedin.metadata.dao.ingestion.preupdate.PreUpdateRoutingClient; | ||
| import com.linkedin.testing.AspectFoo; | ||
|
|
||
|
|
||
| public class SamplePreUpdateRoutingClient implements RestliCompliantPreUpdateRoutingClient { | ||
| @Override | ||
| public Message routingLambda(Message urn, Message aspect) { | ||
| // For testing, change the aspect value to "bar" | ||
| return Any.pack(StringValue.of("bar")); | ||
| } | ||
|
|
||
| @Override | ||
| public Message convertUrnToMessage(Urn urn) { | ||
| // Directly wrap the URN string into a Protobuf message for testing | ||
| return Any.pack(StringValue.of(urn.toString())); | ||
| } | ||
|
|
||
| @Override | ||
| public Message convertAspectToMessage(RecordTemplate pegasusAspect) { | ||
| // For testing, convert AspectFoo to a TestMessageProtos.AspectMessage | ||
| // Assuming the aspect has a `value` field and its string representation can be used for now | ||
| String aspectString = pegasusAspect.toString(); // Extracting the aspect as a string (e.g., {value=foo}) | ||
|
|
||
| // Wrap the aspect string into a simple Protobuf message for testing | ||
| return Any.pack(StringValue.of(aspectString)); | ||
| } | ||
| public class SamplePreUpdateRoutingClient implements PreUpdateRoutingClient { | ||
|
|
||
| @Override | ||
| public RecordTemplate convertAspectToRecordTemplate(Message messageAspect) { | ||
| // For testing, convert TestMessageProtos.AspectMessage back to AspectFoo | ||
| // Create a new RecordTemplate (AspectFoo in this case) and set the value field | ||
| return new AspectFoo().setValue("bar"); | ||
| public PreUpdateResponse preUpdate(Urn urn, RecordTemplate recordTemplate) { | ||
| AspectFoo aspectFoo = (AspectFoo) recordTemplate; | ||
| aspectFoo.setValue("bar"); | ||
| return new PreUpdateResponse(aspectFoo); | ||
| } | ||
| } |
Oops, something went wrong.