@Test
  public void testReplicateBlobOnCompositeBlob() throws Exception {
    try {
      // CompositeBlob with four chunks.
      maxPutChunkSize = PUT_CONTENT_SIZE / 4;
      MockServerLayout layout = new MockServerLayout(mockClusterMap);
      String serviceId = "replicate-blob-service";
      setRouter(getNonBlockingRouterProperties(localDcName), layout, new LoggingNotificationSystem());
      setOperationParams();
      // PutBlob to the three local replicas
      String compositeBlobId =
          router.putBlob(putBlobProperties, putUserMetadata, putChannel, new PutBlobOptionsBuilder().build())
              .get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      GetBlobResult localGetBlobResult =
          router.getBlob(compositeBlobId, new GetBlobOptionsBuilder().build(), null, null).get();
      BlobProperties localProperties = localGetBlobResult.getBlobInfo().getBlobProperties();

      // set error status for all the local replica. read from the remote replica, it should fail
      for (MockServer server : layout.getMockServers()) {
        if (server.getDataCenter().equals(localDcName)) {
          server.setServerErrorForAllRequests(ServerErrorCode.ReplicaUnavailable);
        }
      }

      try {
        router.getBlob(compositeBlobId, new GetBlobOptionsBuilder().build(), null, null).get();
        Assert.fail("Should return Ambry unavailable.");
      } catch (ExecutionException e) {
        Throwable t = e.getCause();
        assertTrue("Cause should be RouterException", t instanceof RouterException);
        assertEquals("ErrorCode mismatch", RouterErrorCode.AmbryUnavailable, ((RouterException) t).getErrorCode());
      }

      layout.getMockServers().forEach(mockServer -> mockServer.setServerErrorForAllRequests(null));

      // Replicate Blob from the local replica to some remote replicas.
      // choose one local replica as the source DataNode. Set the other two replicas as Replica_Unavailable
      DataNodeId sourceDataNode = null;
      for (MockServer server : layout.getMockServers()) {
        if (server.getDataCenter().equals(localDcName)) {
          if (sourceDataNode == null) {
            sourceDataNode = mockClusterMap.getDataNodeId(server.getHostName(), server.getHostPort());
          } else {
            server.setServerErrorForAllRequests(ServerErrorCode.ReplicaUnavailable);
          }
        }
      }
      // since there is only one local available, router will replicate the blob to at least 1 remote replicas.
      // Replicating the four chunks. Also replicate the composite metadata blob.
      GetBlobResult localGetChunkResult = router.getBlob(compositeBlobId,
              new GetBlobOptionsBuilder().operationType(GetBlobOptions.OperationType.BlobChunkIds).build(), null, null)
          .get();
      List<StoreKey> chunkIds = localGetChunkResult.getBlobChunkIds();
      for (StoreKey chunk : chunkIds) {
        router.replicateBlob(chunk.getID(), serviceId, sourceDataNode).get();
      }
      router.replicateBlob(compositeBlobId, serviceId, sourceDataNode).get();

      // set unavailable to all local replicas. So router.getBlob will get the Blob from remote replicas.
      for (MockServer server : layout.getMockServers()) {
        if (server.getDataCenter().equals(localDcName)) {
          server.setServerErrorForAllRequests(ServerErrorCode.ReplicaUnavailable);
        }
      }
      GetBlobResult remoteGetBlobResult =
          router.getBlob(compositeBlobId, new GetBlobOptionsBuilder().build(), null, null).get();

      // verify BlobProperties
      Assert.assertEquals(remoteGetBlobResult.getBlobInfo().getBlobProperties(), localProperties);
      // verify userMetadata
      Assert.assertArrayEquals(remoteGetBlobResult.getBlobInfo().getUserMetadata(), putUserMetadata);
      // verify data content
      RetainingAsyncWritableChannel retainingAsyncWritableChannel = new RetainingAsyncWritableChannel();
      remoteGetBlobResult.getBlobDataChannel().readInto(retainingAsyncWritableChannel, null).get();
      InputStream input = retainingAsyncWritableChannel.consumeContentAsInputStream();
      retainingAsyncWritableChannel.close();
      byte[] content = Utils.readBytesFromStream(input, PUT_CONTENT_SIZE);
      input.close();
      Assert.assertArrayEquals(putContent, content);

      layout.getMockServers().forEach(mockServer -> mockServer.setServerErrorForAllRequests(null));
    } finally {
      // Since we are using same blob ID for all tests, clear cache which stores blobs that are not found in router.
      router.getNotFoundCache().invalidateAll();

      if (router != null) {
        router.close();
      }
    }
  }
Test for ReplicateBlob success case on CompositeBlob
Replicating one Blob to the remote replicas when two local replicas were unavailable.
@throws Exception
ambry-router/src/test/java/com/github/ambry/router/NonBlockingRouterTest.java