Merge branch 'master' of https://github.com/linkedin/ambry into dw-bo…

…otstrap-controller

ambry-api/src/main/java/com/github/ambry/server/StoreManager.java

@@ -19,6 +19,7 @@
 import com.github.ambry.store.Store;
 import com.github.ambry.store.StoreException;
 import java.io.IOException;
+import java.nio.file.FileStore;
 import java.util.Collection;
 import java.util.List;
 import java.util.regex.Pattern;
@@ -36,6 +37,13 @@ public interface StoreManager {
    */
   boolean addBlobStore(ReplicaId replica);
 
+  /**
+   * Add a new FileStore with given {@link ReplicaId}.
+   * @param replicaId the {@link ReplicaId} of the {@link FileStore} which would be added.
+   * @return {@code true} if adding FileStore was successful. {@code false} if not.
+   */
+  boolean addFileStore(ReplicaId replicaId);
+
   /**
    * Remove store from storage manager.
    * @param id the {@link PartitionId} associated with store
@@ -64,6 +72,15 @@ public interface StoreManager {
    */
   Store getStore(PartitionId id);
 
+
+  /**
+   *
+   * @param id the {@link PartitionId} to find the store for.
+   * @return the {@link FileStore} corresponding to the given {@link PartitionId}, or {@code null} if no store was found for
+   *         that partition, or that store was not started.
+   */
+  FileStore getFileStore(PartitionId id);
+
   /**
    * Get replicaId on current node by partition name. (There should be at most one replica belonging to specific
    * partition on single node)

ambry-cloud/src/main/java/com/github/ambry/cloud/CloudStorageManager.java

@@ -23,6 +23,7 @@
 import com.github.ambry.server.ServerErrorCode;
 import com.github.ambry.server.StoreManager;
 import com.github.ambry.store.Store;
+import java.nio.file.FileStore;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
@@ -61,6 +62,16 @@ public boolean addBlobStore(ReplicaId replica) {
     return createAndStartBlobStoreIfAbsent(replica.getPartitionId()) != null;
   }
 
+  /**
+   * Returning false because this will not be used as part of CloudStorageManager Implementation.
+   * Implementation will be added if needed.
+   */
+  @Override
+  public boolean addFileStore(ReplicaId replicaId) {
+    return false;
+
+  }
+
   @Override
   public boolean shutdownBlobStore(PartitionId id) {
     try {
@@ -75,6 +86,10 @@ public boolean shutdownBlobStore(PartitionId id) {
     return false;
   }
 
+  /**
+   * Returning null because this will not be used as part of CloudStorageManager Implementation.
+   * Implementation will be added if needed.
+   */
   @Override
   public Store getStore(PartitionId id) {
     try {
@@ -88,6 +103,11 @@ public Store getStore(PartitionId id) {
     return createAndStartBlobStoreIfAbsent(id);
   }
 
+  @Override
+  public FileStore getFileStore(PartitionId id) {
+    return null;
+  }
+
   @Override
   public boolean scheduleNextForCompaction(PartitionId id) {
     throw new UnsupportedOperationException("Method not supported");

ambry-clustermap/src/main/java/com/github/ambry/clustermap/AmbryReplicaSyncUpManager.java

@@ -65,7 +65,7 @@ public void initiateBootstrap(ReplicaId replicaId) {
 
   @Override
   public void initiateFileCopy(ReplicaId replicaId) {
-    //To BE Filled.
+    //To Be Added With File Copy Protocol
   }
 
   @Override
@@ -108,7 +108,7 @@ public void waitBootstrapCompleted(String partitionName) throws InterruptedExcep
 
   @Override
   public void waitForFileCopyCompleted(String partitionName) throws InterruptedException {
-    //To Be Filled
+    //To Be Added With File Copy Protocol
   }
 
   @Override
@@ -204,7 +204,7 @@ public void onBootstrapComplete(ReplicaId replicaId) {
 
   @Override
   public void onFileCopyComplete(ReplicaId replicaId) {
-    // To Be Filled
+    //To Be Added With File Copy Protocol
   }
 
   @Override

ambry-protocol/src/main/java/com/github/ambry/protocol/FileCopyGetChunkRequest.java

@@ -0,0 +1,99 @@
+/**
+ * Copyright 2024 LinkedIn Corp. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ */
+package com.github.ambry.protocol;
+
+import com.github.ambry.clustermap.ClusterMap;
+import com.github.ambry.clustermap.PartitionId;
+import com.github.ambry.utils.Utils;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.charset.Charset;
+
+
+public class FileCopyGetChunkRequest extends RequestOrResponse{
+  private PartitionId partitionId;
+  private String fileName;
+  private long startOffset;
+  private long chunkLengthInBytes;
+  private static final short File_Chunk_Request_Version_V1 = 1;
+  private static final int File_Name_Size_In_Bytes = 4;
+
+
+  public FileCopyGetChunkRequest( short versionId, int correlationId,
+      String clientId, PartitionId partitionId, String fileName, long startOffset, long sizeInBytes) {
+    super(RequestOrResponseType.FileCopyGetChunkRequest, versionId, correlationId, clientId);
+    if(partitionId == null || fileName.isEmpty() || startOffset < 0 || sizeInBytes < 0){
+      throw new IllegalArgumentException("PartitionId, FileName, StartOffset or SizeInBytes cannot be null or negative");
+    }
+    this.partitionId = partitionId;
+    this.fileName = fileName;
+    this.startOffset = startOffset;
+    this.chunkLengthInBytes = sizeInBytes;
+  }
+
+  public static FileCopyGetChunkRequest readFrom(DataInputStream stream, ClusterMap clusterMap)
+      throws IOException {
+    Short versionId = stream.readShort();
+    validateVersion(versionId);
+    int correlationId = stream.readInt();
+    String clientId = Utils.readIntString(stream);
+    PartitionId partitionId = clusterMap.getPartitionIdFromStream(stream);
+    String fileName = Utils.readIntString(stream);
+    long startOffset = stream.readLong();
+    long sizeInBytes = stream.readLong();
+    return new FileCopyGetChunkRequest(versionId, correlationId, clientId, partitionId, fileName, startOffset, sizeInBytes);
+  }
+
+  protected void prepareBuffer(){
+    super.prepareBuffer();
+    bufferToSend.writeBytes(partitionId.getBytes());
+    Utils.serializeString(bufferToSend, fileName, Charset.defaultCharset());
+    bufferToSend.writeLong(startOffset);
+    bufferToSend.writeLong(chunkLengthInBytes);
+  }
+
+  public String toString(){
+    StringBuilder sb = new StringBuilder();
+    sb.append("FileCopyProtocolGetChunkRequest[")
+      .append("PartitionId=").append(partitionId)
+      .append(", FileName=").append(fileName)
+      .append(", StartOffset=").append(startOffset)
+      .append(", SizeInBytes=").append(chunkLengthInBytes)
+      .append("]");
+    return sb.toString();
+  }
+
+  public long sizeInBytes() {
+    return super.sizeInBytes() + partitionId.getBytes().length + File_Name_Size_In_Bytes + fileName.length() + Long.BYTES + Long.BYTES;
+  }
+
+  public PartitionId getPartitionId() {
+    return partitionId;
+  }
+  public String getFileName() {
+    return fileName;
+  }
+  public long getStartOffset() {
+    return startOffset;
+  }
+  public long getChunkLengthInBytes() {
+    return chunkLengthInBytes;
+  }
+
+  static void validateVersion(short version){
+    if (version != File_Chunk_Request_Version_V1) {
+      throw new IllegalArgumentException("Unknown version for FileMetadataRequest: " + version);
+    }
+  }
+}

ambry-protocol/src/main/java/com/github/ambry/protocol/FileCopyGetMetaDataRequest.java

@@ -0,0 +1,82 @@
+/**
+ * Copyright 2024 LinkedIn Corp. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ */
+package com.github.ambry.protocol;
+
+import com.github.ambry.clustermap.ClusterMap;
+import com.github.ambry.clustermap.PartitionId;
+import com.github.ambry.utils.Utils;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.charset.Charset;
+
+public class FileCopyGetMetaDataRequest extends RequestOrResponse{
+  private PartitionId partitionId;
+  private String hostName;
+  private static final short File_Metadata_Request_Version_V1 = 1;
+  private static final int HostName_Field_Size_In_Bytes = 4;
+
+  public FileCopyGetMetaDataRequest(short versionId, int correlationId, String clientId,
+      PartitionId partitionId, String hostName) {
+    super(RequestOrResponseType.FileCopyGetMetaDataRequest, versionId, correlationId, clientId);
+    if (partitionId == null) {
+      throw new IllegalArgumentException("Partition cannot be null");
+    }
+    if (hostName.isEmpty()){
+      throw new IllegalArgumentException("Host Name cannot be null");
+    }
+    this.partitionId = partitionId;
+    this.hostName = hostName;
+  }
+
+  public String getHostName() {
+    return hostName;
+  }
+
+  public PartitionId getPartitionId() {
+    return partitionId;
+  }
+
+  protected static FileCopyGetMetaDataRequest readFrom(DataInputStream stream, ClusterMap clusterMap) throws IOException {
+    Short versionId = stream.readShort();
+    validateVersion(versionId);
+    int correlationId = stream.readInt();
+    String clientId = Utils.readIntString(stream);
+    String hostName = Utils.readIntString(stream);
+    PartitionId partitionId = clusterMap.getPartitionIdFromStream(stream);
+    return new FileCopyGetMetaDataRequest(versionId, correlationId, clientId, partitionId, hostName);
+  }
+
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    sb.append("FileMetaDataRequest[").append("PartitionId=").append(partitionId).append(", HostName=").append(hostName)
+        .append("]");
+    return sb.toString();
+  }
+
+  public long sizeInBytes() {
+    return super.sizeInBytes() + HostName_Field_Size_In_Bytes + hostName.length() + partitionId.getBytes().length;
+  }
+
+  protected void prepareBuffer() {
+    super.prepareBuffer();
+    Utils.serializeString(bufferToSend, hostName, Charset.defaultCharset());
+    bufferToSend.writeBytes(partitionId.getBytes());
+  }
+
+  static void validateVersion(short version) {
+    if (version != File_Metadata_Request_Version_V1) {
+      throw new IllegalArgumentException("Unknown version for FileMetadataRequest: " + version);
+    }
+  }
+}

ambry-protocol/src/main/java/com/github/ambry/protocol/FileCopyGetMetaDataResponse.java

@@ -0,0 +1,93 @@
+/**
+ * Copyright 2024 LinkedIn Corp. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ */
+package com.github.ambry.protocol;
+
+import com.github.ambry.server.ServerErrorCode;
+import com.github.ambry.utils.Utils;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
+
+
+public class FileCopyGetMetaDataResponse extends Response {
+  private final int numberOfLogfiles;
+  private final List<LogInfo> logInfos;
+  private static final short File_Copy_Protocol_Metadata_Response_Version_V1 = 1;
+
+  public FileCopyGetMetaDataResponse(short versionId, int correlationId, String clientId, int numberOfLogfiles,
+      List<LogInfo> logInfos, ServerErrorCode errorCode) {
+    super(RequestOrResponseType.FileCopyGetMetaDataResponse, versionId, correlationId, clientId, errorCode);
+    validateVersion(versionId);
+    this.numberOfLogfiles = numberOfLogfiles;
+    this.logInfos = logInfos;
+  }
+
+  public static FileCopyGetMetaDataResponse readFrom(DataInputStream stream) throws IOException {
+    RequestOrResponseType type = RequestOrResponseType.values()[stream.readShort()];
+    if (type != RequestOrResponseType.FileCopyGetMetaDataResponse) {
+      throw new IllegalArgumentException("The type of request response is not compatible. Expected : {}, Actual : {}" +
+          RequestOrResponseType.FileCopyGetMetaDataResponse + type);
+    }
+    short versionId = stream.readShort();
+    int correlationId = stream.readInt();
+    String clientId = Utils.readIntString(stream);
+    ServerErrorCode errorCode = ServerErrorCode.values()[stream.readShort()];
+
+    if(errorCode != ServerErrorCode.No_Error) {
+      //Setting the number of logfiles to 0 as there are no logfiles to be read.
+      return new FileCopyGetMetaDataResponse(versionId, correlationId, clientId, 0, new ArrayList<>(), errorCode);
+    }
+
+    int numberOfLogfiles = stream.readInt();
+    List<LogInfo> logInfos = new ArrayList<>();
+    for (int i = 0; i < numberOfLogfiles; i++) {
+      logInfos.add(LogInfo.readFrom(stream));
+    }
+    return new FileCopyGetMetaDataResponse(versionId, correlationId, clientId, numberOfLogfiles, logInfos, errorCode);
+  }
+  protected void prepareBuffer() {
+    super.prepareBuffer();
+    bufferToSend.writeInt(numberOfLogfiles);
+    for (LogInfo logInfo : logInfos) {
+      logInfo.writeTo(bufferToSend);
+    }
+  }
+
+  public long sizeInBytes() {
+    return super.sizeInBytes() + Integer.BYTES + logInfos.stream().mapToLong(LogInfo::sizeInBytes).sum();
+  }
+
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    sb.append("FileMetaDataResponse[NumberOfLogfiles=").append(numberOfLogfiles).append(", logInfoList").append(
+        logInfos.toString()).append("]");
+    return sb.toString();
+  }
+
+  public int getNumberOfLogfiles() {
+    return numberOfLogfiles;
+  }
+
+  public List<LogInfo> getLogInfos() {
+    return logInfos;
+  }
+
+  static void validateVersion(short version) {
+    if (version != File_Copy_Protocol_Metadata_Response_Version_V1) {
+      throw new IllegalArgumentException("Unknown version for FileCopyProtocolMetaDataResponse: " + version);
+    }
+  }
+}