-
Notifications
You must be signed in to change notification settings - Fork 79
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'linkedin:main' into kristy_lee/650
- Loading branch information
Showing
240 changed files
with
8,241 additions
and
1,841 deletions.
There are no files selected for viewing
Large diffs are not rendered by default.
Oops, something went wrong.
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
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
2 changes: 1 addition & 1 deletion
2
...ice/blobtransfer/BlobTransferManager.java → ...nci/blobtransfer/BlobTransferManager.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
66 changes: 66 additions & 0 deletions
66
...client/src/main/java/com/linkedin/davinci/blobtransfer/BlobTransferPartitionMetadata.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,66 @@ | ||
| package com.linkedin.davinci.blobtransfer; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
|
|
||
|
|
||
| /** | ||
| * This class is the metadata of a partition in the blob transfer client | ||
| */ | ||
| public class BlobTransferPartitionMetadata { | ||
| public String topicName; | ||
| public int partitionId; | ||
| public java.nio.ByteBuffer offsetRecord; | ||
| public java.nio.ByteBuffer storeVersionState; | ||
|
|
||
| public BlobTransferPartitionMetadata() { | ||
| } | ||
|
|
||
| public BlobTransferPartitionMetadata( | ||
| @JsonProperty("topicName") String topicName, | ||
| @JsonProperty("partitionId") int partitionId, | ||
| @JsonProperty("offsetRecord") java.nio.ByteBuffer offsetRecord, | ||
| @JsonProperty("storeVersionState") java.nio.ByteBuffer storeVersionState) { | ||
| this.topicName = topicName; | ||
| this.partitionId = partitionId; | ||
| this.offsetRecord = offsetRecord; | ||
| this.storeVersionState = storeVersionState; | ||
| } | ||
|
|
||
| public String getTopicName() { | ||
| return topicName; | ||
| } | ||
|
|
||
| public void setTopicName(String topicName) { | ||
| this.topicName = topicName; | ||
| } | ||
|
|
||
| public int getPartitionId() { | ||
| return partitionId; | ||
| } | ||
|
|
||
| public void setPartitionId(int partitionId) { | ||
| this.partitionId = partitionId; | ||
| } | ||
|
|
||
| public java.nio.ByteBuffer getOffsetRecord() { | ||
| return offsetRecord; | ||
| } | ||
|
|
||
| public void setOffsetRecord(java.nio.ByteBuffer offsetRecord) { | ||
| this.offsetRecord = offsetRecord; | ||
| } | ||
|
|
||
| public void setStoreVersionState(java.nio.ByteBuffer storeVersionState) { | ||
| this.storeVersionState = storeVersionState; | ||
| } | ||
|
|
||
| public java.nio.ByteBuffer getStoreVersionState() { | ||
| return storeVersionState; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "BlobTransferPartitionMetadata {" + " topicName='" + topicName + ", partitionId=" + partitionId | ||
| + ", offsetRecord=" + offsetRecord + ", storeVersionState=" + storeVersionState + " }"; | ||
| } | ||
| } |
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
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
27 changes: 27 additions & 0 deletions
27
...ts/da-vinci-client/src/main/java/com/linkedin/davinci/blobtransfer/BlobTransferUtils.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,27 @@ | ||
| package com.linkedin.davinci.blobtransfer; | ||
|
|
||
| import io.netty.handler.codec.http.HttpResponse; | ||
|
|
||
|
|
||
| public class BlobTransferUtils { | ||
| public static final String BLOB_TRANSFER_STATUS = "X-Blob-Transfer-Status"; | ||
| public static final String BLOB_TRANSFER_COMPLETED = "Completed"; | ||
| public static final String BLOB_TRANSFER_TYPE = "X-Blob-Transfer-Type"; | ||
|
|
||
| public enum BlobTransferType { | ||
| FILE, METADATA | ||
| } | ||
|
|
||
| /** | ||
| * Check if the HttpResponse message is for metadata. | ||
| * @param msg the HttpResponse message | ||
| * @return true if the message is a metadata message, false otherwise | ||
| */ | ||
| public static boolean isMetadataMessage(HttpResponse msg) { | ||
| String metadataHeader = msg.headers().get(BlobTransferUtils.BLOB_TRANSFER_TYPE); | ||
| if (metadataHeader == null) { | ||
| return false; | ||
| } | ||
| return metadataHeader.equals(BlobTransferUtils.BlobTransferType.METADATA.name()); | ||
| } | ||
| } |
8 changes: 5 additions & 3 deletions
8
...transfer/NettyP2PBlobTransferManager.java → ...transfer/NettyP2PBlobTransferManager.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
2 changes: 1 addition & 1 deletion
2
.../blobtransfer/P2PBlobTransferManager.java → .../blobtransfer/P2PBlobTransferManager.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
30 changes: 30 additions & 0 deletions
30
...nci-client/src/main/java/com/linkedin/davinci/blobtransfer/client/MetadataAggregator.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,30 @@ | ||
| package com.linkedin.davinci.blobtransfer.client; | ||
|
|
||
| import com.linkedin.davinci.blobtransfer.BlobTransferUtils; | ||
| import io.netty.handler.codec.http.HttpObjectAggregator; | ||
| import io.netty.handler.codec.http.HttpResponse; | ||
|
|
||
|
|
||
| /** | ||
| * MetadataAggregator is a custom HttpObjectAggregator that | ||
| * only aggregated HttpResponse messages for metadata. | ||
| */ | ||
| public class MetadataAggregator extends HttpObjectAggregator { | ||
| public MetadataAggregator(int maxContentLength) { | ||
| super(maxContentLength); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean acceptInboundMessage(Object msg) throws Exception { | ||
| if (msg instanceof HttpResponse) { | ||
| HttpResponse httpMessage = (HttpResponse) msg; | ||
| // only accept metadata messages to be aggregated | ||
| if (BlobTransferUtils.isMetadataMessage(httpMessage)) { | ||
| return super.acceptInboundMessage(msg); | ||
| } else { | ||
| return false; | ||
| } | ||
| } | ||
| return super.acceptInboundMessage(msg); | ||
| } | ||
| } |
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
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
Oops, something went wrong.