Remove cluster name for blobId used in router (#2913)

* Remove cluster name for blobId used in router

* Adding tests

* Adding unit test

---------

Co-authored-by: Sophie Guo <sopguo@sopguo-mn2.linkedin.biz>

ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobPutHandler.java

@@ -249,7 +249,7 @@ private Callback<String> routerPutBlobCallback(BlobInfo blobInfo) {
       return buildCallback(frontendMetrics.putRouterPutBlobMetrics, blobId -> {
         restResponseChannel.setHeader(RestUtils.Headers.BLOB_SIZE, restRequest.getBlobBytesReceived());
         restResponseChannel.setHeader(RestUtils.Headers.LOCATION, blobId);
-        String blobIdClean = RestUtils.stripSlashAndExtensionFromId(blobId);
+        String blobIdClean = stripPrefixAndExtension(blobId);
         if (blobInfo.getBlobProperties().getTimeToLiveInSeconds() == Utils.Infinite_Time) {
           // Do ttl update with retryExecutor. Use the blob ID returned from the router instead of the converted ID
           // since the converted ID may be changed by the ID converter.
@@ -767,4 +767,10 @@ private <T> Callback<T> deleteDatasetVersionIfUploadFailedCallBack(Callback<T> c
       };
     }
   }
+
+  public String stripPrefixAndExtension(String blobId) throws RestServiceException {
+    return RestUtils.stripSlashAndExtensionFromId(
+        RequestPath.parse(blobId, Collections.emptyMap(), frontendConfig.pathPrefixesToRemove, clusterName)
+            .getOperationOrBlobId(false));
+  }
 }

ambry-frontend/src/test/java/com/github/ambry/frontend/FrontendRestRequestServiceTest.java

@@ -321,6 +321,49 @@ public void useServiceWithoutStartTest() throws Exception {
     }
   }
 
+  @Test
+  public void testNamedBlobPut() throws Exception {
+    Account testAccount = new ArrayList<>(accountService.getAllAccounts()).get(1);
+    Container testContainer = new ArrayList<>(testAccount.getAllContainers()).get(1);
+    String blobName = "blobName";
+    String namedBlobPathUri =
+        NAMED_BLOB_PREFIX + SLASH + testAccount.getName() + SLASH + testContainer.getName() + SLASH + blobName;
+    ByteBuffer content = ByteBuffer.wrap(TestUtils.getRandomBytes(10));
+    List<ByteBuffer> body = new LinkedList<>();
+    body = new LinkedList<>();
+    body.add(content);
+    body.add(null);
+    JSONObject headers = new JSONObject().put(RestUtils.Headers.TARGET_ACCOUNT_NAME, testAccount.getName())
+        .put(RestUtils.Headers.TARGET_CONTAINER_NAME, testContainer.getName());
+    setAmbryHeadersForPut(headers, -1, testContainer.isCacheable(), "test", "application/octet-stream", "owner", null,
+        null, null);
+    RestRequest restRequest = createRestRequest(RestMethod.PUT, namedBlobPathUri, headers, body);
+    BlobProperties blobProperties =
+        new BlobProperties(0, testAccount.getName(), "owner", "image/gif", false, 7200, testAccount.getId(),
+            testContainer.getId(), false, null, null, null);
+    ReadableStreamChannel byteBufferContent = new ByteBufferReadableStreamChannel(ByteBuffer.allocate(10));
+    String blobIdFromRouter =
+        router.putBlobWithIdVersion(blobProperties, new byte[0], byteBufferContent, BlobId.BLOB_ID_V6).get();
+    String blobIdWithClusterName = "/" + CLUSTER_NAME + "/" + blobIdFromRouter + ".bin";
+    reset(namedBlobDb);
+    NamedBlobRecord namedBlobRecord =
+        new NamedBlobRecord(testAccount.getName(), testContainer.getName(), blobName, blobIdWithClusterName, 3600);
+    NamedBlobRecord namedBlobRecordAfterTtlUpdate =
+        new NamedBlobRecord(testAccount.getName(), testContainer.getName(), blobName, blobIdWithClusterName,
+            Utils.Infinite_Time);
+    when(namedBlobDb.put(any(), any(), any())).thenReturn(
+        CompletableFuture.completedFuture(new PutResult(namedBlobRecord)));
+    when(namedBlobDb.get(namedBlobRecord.getAccountName(), namedBlobRecord.getContainerName(), blobName,
+        GetOption.None)).thenReturn(CompletableFuture.completedFuture(namedBlobRecord));
+    when(namedBlobDb.updateBlobTtlAndStateToReady(any())).thenReturn(
+        CompletableFuture.completedFuture(new PutResult(namedBlobRecordAfterTtlUpdate)));
+
+    MockRestResponseChannel restResponseChannel = new MockRestResponseChannel();
+    doOperation(restRequest, restResponseChannel);
+    assertEquals("Unexpected response status", ResponseStatus.Created, restResponseChannel.getStatus());
+    assertEquals("Unexpected blob Id", blobIdWithClusterName, restResponseChannel.getHeader(LOCATION));
+  }
+
   /**
    * Checks for reactions of all methods in {@link FrontendRestRequestService} to null arguments.
    * @throws Exception

ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobPutHandlerTest.java

@@ -92,6 +92,7 @@ public class NamedBlobPutHandlerTest {
   private static final String CONVERTED_ID = "/abcdef";
   private static final String CLUSTER_NAME = "ambry-test";
   private static final String NAMED_BLOB_PREFIX = "/named";
+  private static final String PATH_PREFIX_TO_REMOVE = "/media";
   private static final String SLASH = "/";
   private static final String BLOBNAME = "ambry_blob_name";
   private static final String DATASET_NAME = "testDataset";
@@ -192,6 +193,27 @@ public void updateNamedBlobAllowedTest() throws Exception {
     assertEquals("Unexpected response status", restResponseChannel.getStatus(), ResponseStatus.Ok);
   }
 
+
+  @Test
+  public void testStripPrefixAndExtension() throws Exception {
+    String EXTENSION = ".bin";
+    Properties properties = new Properties();
+    properties.setProperty("frontend.path.prefixes.to.remove",  PATH_PREFIX_TO_REMOVE);
+    initNamedBlobPutHandler(properties);
+
+    String blobId = CLUSTER_NAME + "/" + CONVERTED_ID + EXTENSION;
+    assertEquals("Blob Id should not have prefix and extention", CONVERTED_ID,
+        namedBlobPutHandler.stripPrefixAndExtension(blobId));
+
+    blobId = PATH_PREFIX_TO_REMOVE + "/" + CLUSTER_NAME + "/" + CONVERTED_ID + EXTENSION;
+    assertEquals("Blob Id should not have prefix and extention", CONVERTED_ID,
+        namedBlobPutHandler.stripPrefixAndExtension(blobId));
+
+    blobId = "/" + CONVERTED_ID + EXTENSION;
+    assertEquals("Blob Id should not have prefix and extention", CONVERTED_ID,
+        namedBlobPutHandler.stripPrefixAndExtension(blobId));
+  }
+
   /**
    * Do NOT allow upsert for container with NamedBlobMode = NO_UPDATE
    */