@Test
  public void testReplicateBlobV2() throws Exception {
    Assume.assumeTrue(
        MessageFormatRecord.getCurrentMessageHeaderVersion() >= MessageFormatRecord.Message_Header_Version_V3);
    Map<DataNodeId, MockHost> hosts = new HashMap<>();
    MockPartitionId partitionId =
        (MockPartitionId) clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS).get(0);
    MockHost sourceHost = null; // the remote source host to replicate the blob from
    for (ReplicaId replica : partitionId.getReplicaIds()) {
      MockHost host = new MockHost(replica.getDataNodeId(), clusterMap);
      if (dataNodeId != host.dataNodeId) {
        if (sourceHost == null) {
          sourceHost = host;
        }
      }
      hosts.put(host.dataNodeId, host);
    }

    MockConnectionPool connectionPool = new MockConnectionPool(hosts, clusterMap, 5);
    ambryRequests = new AmbryServerRequests(storageManager, requestResponseChannel, clusterMap, dataNodeId,
        clusterMap.getMetricRegistry(), serverMetrics, findTokenHelper, null, replicationManager, storeKeyFactory,
        serverConfig, diskManagerConfig, storeKeyConverterFactory, statsManager, helixParticipant, connectionPool);
    int correlationId = TestUtils.RANDOM.nextInt();
    String clientId = TestUtils.getRandomString(10);
    storageManager.resetStore();

    // PutBlob and TtlUpdate exists on the source host
    BlobId blobId = (BlobId) addPutMessagesToReplicasOfPartition(partitionId, Arrays.asList(sourceHost), 1).get(0);
    addTtlUpdateMessagesToReplicasOfPartition(partitionId, (StoreKey) blobId, Arrays.asList(sourceHost), -1);
    ReplicateBlobRequest request =
        new ReplicateBlobRequest(correlationId, clientId, blobId, sourceHost.dataNodeId.getHostname(),
            sourceHost.dataNodeId.getPort(), RequestOrResponseType.PutRequest, 0, (short) 0, 0);
    assertEquals(ReplicateBlobRequest.VERSION_2, request.getVersionId());
    sendRequestGetResponse(request, ServerErrorCode.BadRequest);

    storageManager.resetStore();
    request = new ReplicateBlobRequest(correlationId, clientId, blobId, sourceHost.dataNodeId.getHostname(),
        sourceHost.dataNodeId.getPort(), RequestOrResponseType.TtlUpdateRequest, 0, (short) -1, 0);
    assertEquals(ReplicateBlobRequest.VERSION_2, request.getVersionId());
    validKeysInStore.add(blobId);
    sendRequestGetResponse(request, ServerErrorCode.NoError);
    assertEquals("Received operation is not expected.", RequestOrResponseType.TtlUpdateRequest,
        MockStorageManager.operationReceived);
    validKeysInStore.remove(blobId);

    // PutBlob, TtlUpdate and Delete exists on the source host
    blobId = (BlobId) addPutMessagesToReplicasOfPartition(partitionId, Arrays.asList(sourceHost), 1).get(0);
    addTtlUpdateMessagesToReplicasOfPartition(partitionId, (StoreKey) blobId, Arrays.asList(sourceHost), -1);
    addDeleteMessagesToReplicasOfPartition(partitionId, (StoreKey) blobId, Arrays.asList(sourceHost));
    request = new ReplicateBlobRequest(correlationId, clientId, blobId, sourceHost.dataNodeId.getHostname(),
        sourceHost.dataNodeId.getPort(), RequestOrResponseType.PutRequest, 0, (short) 0, 0);
    assertEquals(ReplicateBlobRequest.VERSION_2, request.getVersionId());
    sendRequestGetResponse(request, ServerErrorCode.BadRequest);

    request = new ReplicateBlobRequest(correlationId, clientId, blobId, sourceHost.dataNodeId.getHostname(),
        sourceHost.dataNodeId.getPort(), RequestOrResponseType.TtlUpdateRequest, 0, (short) -1, 0);
    assertEquals(ReplicateBlobRequest.VERSION_2, request.getVersionId());
    validKeysInStore.add(blobId);
    sendRequestGetResponse(request, ServerErrorCode.NoError);
    assertEquals("Received operation is not expected.", RequestOrResponseType.TtlUpdateRequest,
        MockStorageManager.operationReceived);
    validKeysInStore.remove(blobId);

    request = new ReplicateBlobRequest(correlationId, clientId, blobId, sourceHost.dataNodeId.getHostname(),
        sourceHost.dataNodeId.getPort(), RequestOrResponseType.DeleteRequest, 0, (short) -1, 0);
    assertEquals(ReplicateBlobRequest.VERSION_2, request.getVersionId());
    validKeysInStore.add(blobId);
    sendRequestGetResponse(request, ServerErrorCode.NoError);
    assertEquals("Received operation is not expected.", RequestOrResponseType.DeleteRequest,
        MockStorageManager.operationReceived);
    validKeysInStore.remove(blobId);

    // PutBlob and Delete exists on the source host
    blobId = (BlobId) addPutMessagesToReplicasOfPartition(partitionId, Arrays.asList(sourceHost), 1).get(0);
    addDeleteMessagesToReplicasOfPartition(partitionId, (StoreKey) blobId, Arrays.asList(sourceHost));
    request = new ReplicateBlobRequest(correlationId, clientId, blobId, sourceHost.dataNodeId.getHostname(),
        sourceHost.dataNodeId.getPort(), RequestOrResponseType.DeleteRequest, 0, (short) -1, 0);
    assertEquals(ReplicateBlobRequest.VERSION_2, request.getVersionId());
    validKeysInStore.add(blobId);
    sendRequestGetResponse(request, ServerErrorCode.NoError);
    assertEquals("Received operation is not expected.", RequestOrResponseType.DeleteRequest,
        MockStorageManager.operationReceived);
    validKeysInStore.remove(blobId);

    // We test the cases that target under different conditions with integration test.
  }
Tests ReplicateBlobRequestV2 to repair TtlUpdate or Delete.
ambry-server/src/test/java/com/github/ambry/server/AmbryServerRequestsTest.java