static void replicateBlobV2CaseTest(MockCluster cluster, SSLConfig clientSSLConfig, boolean testEncryption,
      MockNotificationSystem notificationSystem) {
    List<MockDataNodeId> allNodes = cluster.getAllDataNodes();
    MockClusterMap clusterMap = cluster.getClusterMap();
    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS);
    PartitionId partitionId = partitionIds.get(0);

    ArrayList<String> dataCenterList = new ArrayList<>(Arrays.asList("DC1", "DC2", "DC3"));
    List<DataNodeId> dataNodes = cluster.getOneDataNodeFromEachDatacenter(dataCenterList);
    /*
     * sourceDataNode is the data node we send PutBlob, TtlUpdate, Delete requests.
     * targetDataNode is the data node we run ReplicateBlob. targetDataNode replicates Blobs from the sourceDataNode.
     * thirdDataNode uses the regular replication thread to replicate the Blobs from peer replicas.
     */
    DataNodeId sourceDataNode = dataNodes.get(0);
    DataNodeId targetDataNode = dataNodes.get(1);
    DataNodeId thirdDataNode = dataNodes.get(2);

    try {
      // Stop replication for this partition on all the servers.
      controlReplicationForPartition(allNodes, partitionId, clientSSLConfig, null, false);

      BlobIdFactory blobIdFactory = new BlobIdFactory(clusterMap);
      short dataSize = 31870;
      byte[] userMetadata = new byte[1000];
      byte[] data = new byte[dataSize];
      byte[] encryptionKey = new byte[100];
      short accountId = Utils.getRandomShort(TestUtils.RANDOM);
      short containerId = Utils.getRandomShort(TestUtils.RANDOM);
      TestUtils.RANDOM.nextBytes(userMetadata);
      TestUtils.RANDOM.nextBytes(data);
      TestUtils.RANDOM.nextBytes(encryptionKey);

      // property with 7days expiration
      BlobProperties propertiesWithTtl =
          new BlobProperties(dataSize, "serviceid1", "ownerid", "image/png", false, TestUtils.TTL_SECS,
              cluster.time.milliseconds(), accountId, containerId, testEncryption, null, null, null, null);
      long ttlUpdateBlobExpiryTimeMs = getExpiryTimeMs(propertiesWithTtl);
      // property of blob with the infinite TTL
      BlobProperties properties = new BlobProperties(dataSize, "serviceid1", accountId, containerId, testEncryption,
          cluster.time.milliseconds());
      cluster.time.sleep(1000);

      short blobIdVersion = CommonTestUtils.getCurrentBlobIdVersion();
      BlobId blobId0 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId1 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId2 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId3 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId4 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId5 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId6 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId7 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId8 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId9 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId10 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId11 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId12 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId13 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      BlobId blobId14 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);

      Port sourcePort = new Port(sourceDataNode.getHttp2Port(), PortType.HTTP2);
      Port targetPort = new Port(targetDataNode.getHttp2Port(), PortType.HTTP2);
      Port thirdPort = new Port(thirdDataNode.getHttp2Port(), PortType.HTTP2);
      ConnectedChannel sourceChannel =
          getBlockingChannelBasedOnPortType(sourcePort, "localhost", null, clientSSLConfig);
      ConnectedChannel targetChannel =
          getBlockingChannelBasedOnPortType(targetPort, "localhost", null, clientSSLConfig);
      ConnectedChannel thirdChannel = getBlockingChannelBasedOnPortType(thirdPort, "localhost", null, clientSSLConfig);
      sourceChannel.connect();
      targetChannel.connect();
      thirdChannel.connect();

      BlobId blobId;
      long operationTime;
      PutRequest putRequest;

      /*
       * Stage 0: Verify the ODP's notification system
       */
      // On the sourceDataNode putBlob and TtlUpdate. On the targetDataNode, blob doesn't exist.
      // BlobId0 is only used to test notification System to verify ODP notification works.
      blobId = blobId0;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      operationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, blobId, operationTime);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // replicateType is RequestOrResponse.TtlUpdateRequest, it replicates PutBlob and then repair the TtlUpdate
      final BlobId asyncBlob = blobId;
      final long asyncOperationTime = operationTime;
      Runnable asyncODP = () -> {
        try {
          Thread.sleep(2000);
          replicateTtlUpdate(targetChannel, asyncBlob, sourceDataNode, asyncOperationTime, Utils.Infinite_Time,
              ServerErrorCode.NoError);
        } catch (InterruptedException e) {

        } catch (IOException e) {
          assertTrue("ReplicateTtlUpdate failure " + e.getMessage(), false);
        }
      };
      new Thread(asyncODP).start();
      // wait for targetDataNode to finish
      notificationSystem.awaitBlobReplicates(blobId.getID());
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // the TtlUpdate Operation Time will be same as the PutBlob time since we apply TtlUpdate from the GetRequest.
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, propertiesWithTtl.getCreationTimeInMs(),
          Utils.Infinite_Time);
      cluster.time.sleep(1000);

      /*
       * Stage 1: Test the correctness of the ReplicateBlob. Replicate blobs from sourceDataNode to the targetDataNode
       * On the sourceDataNode and the targetDataNode, the blob is under different states.
       */
      //
      // sourceDataNode has PutBlob
      //
      // 1. On the sourceDataNode, putBlob. On the targetDataNode, blob doesn't exist. Replicate PutBlob
      blobId = blobId1;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      // replicateType is RequestOrResponse.PutRequest, return Bad_Request
      replicatePutBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.BadRequest);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      cluster.time.sleep(1000);

      // 2. On the sourceDataNode, putBlob. On the thirdDataNode, it has the PutBlob and TtlUpdate.
      // On the targetDataNode, blob doesn't exist. Replicate TtlUpdate even source only has the PutBlob
      blobId = blobId2;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, thirdChannel, ServerErrorCode.NoError);
      operationTime = cluster.time.milliseconds();
      updateBlobTtl(thirdChannel, blobId, operationTime);
      checkTtlUpdateStatus(thirdChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // replicateType is RequestOrResponse.TtlUpdateRequest, it replicates PutBlob and then repair the TtlUpdate
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, Utils.Infinite_Time,
          ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // cannot verify the TtlUpdate operationTime or lifeVersion with GetBlob. GetBlob returns the creation time.
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, operationTime, Utils.Infinite_Time);
      cluster.time.sleep(1000);

      // 3. On the sourceDataNode, putBlob. On the thirdDataNode, putBlob and DeleteBlob
      // On the targetDataNode, blob doesn't exist. Repair DeleteRecord even sourceDataNode doesn't have DeleteRecord.
      blobId = blobId3;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, thirdChannel, ServerErrorCode.NoError);
      operationTime = cluster.time.milliseconds();
      deleteBlob(thirdChannel, blobId, operationTime, ServerErrorCode.NoError);
      // replicateType is RequestOrResponse.DeleteRequest. It repair the PutRecord and the DeleteRecord.
      replicateDeleteBlob(targetChannel, blobId, sourceDataNode, operationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
      checkDeleteRecord(targetChannel, clusterMap, blobId, operationTime);
      cluster.time.sleep(1000);

      // 4. On the sourceDataNode, putBlob. ThirdDataNode has DeleteRecord.
      // On the targetDataNode, blob doesn't exist. Replicate TtlUpdate even source only has the PutBlob.
      blobId = blobId4;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, thirdChannel, ServerErrorCode.NoError);
      operationTime = cluster.time.milliseconds();
      deleteBlob(thirdChannel, blobId, operationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, thirdChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
      // replicateType is RequestOrResponse.TtlUpdateRequest, it replicates PutBlob and then repair the TtlUpdate
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, Utils.Infinite_Time,
          ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, operationTime, Utils.Infinite_Time);
      cluster.time.sleep(1000);

      // 5. On the sourceDataNode, putBlob. ThirdDataNode has PutBlob and TtlUpdate
      // On the targetDataNode, blob doesn't exist. Repair DeleteRecord even sourceData doesn't have DeleteRecord.
      blobId = blobId5;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, thirdChannel, ServerErrorCode.NoError);
      operationTime = cluster.time.milliseconds();
      updateBlobTtl(thirdChannel, blobId, operationTime);
      checkTtlUpdateStatus(thirdChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // replicateType is RequestOrResponse.DeleteRequest. It repair the PutBlob and DeleteRecord.
      replicateDeleteBlob(targetChannel, blobId, sourceDataNode, operationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
      checkDeleteRecord(targetChannel, clusterMap, blobId, operationTime);
      cluster.time.sleep(1000);

      //
      // sourceDataNode has PutBlob and TtlUpdate
      //
      // 6. On the sourceDataNode, the blob is ttlUpdated. On the targetDataNode, it doesn't exist.
      // ReplicateBlob will replicate both the putBlob and the TtlUpdate
      blobId = blobId6;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, false, ttlUpdateBlobExpiryTimeMs);
      operationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, blobId, operationTime);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // replicate to the targetDataNode
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, -1, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // the TtlUpdate Operation Time will be same as the PutBlob time since we apply TtlUpdate from the GetRequest.
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, propertiesWithTtl.getCreationTimeInMs(),
          Utils.Infinite_Time);
      cluster.time.sleep(1000);

      // 7. On the sourceDatanode blob is ttlUpdated. On the targetDataNode, putBlob.
      // Run different Repair Types
      blobId = blobId7;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      operationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, blobId, operationTime);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      // replicateType is RequestOrResponse.PutRequest, Bad_Request
      replicatePutBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.BadRequest);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, false, ttlUpdateBlobExpiryTimeMs);
      // replicateType is RequestOrResponseType.TtlUpdateRequest, so will replicate the TtlUpdate.
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, Utils.Infinite_Time,
          ServerErrorCode.NoError);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, operationTime, Utils.Infinite_Time);
      cluster.time.sleep(1000);

      // 8. On the sourceDatanode blob is ttlUpdated. On the targetDataNode, the blob is also ttlUpdated already.
      // Run three different Repair Types
      blobId = blobId8;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      operationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, blobId, operationTime);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      updateBlobTtl(targetChannel, blobId, operationTime);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // replicateType is RequestOrResponse.PutRequest, do nothing
      replicatePutBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.BadRequest);
      // replicateType is RequestOrResponseType.TtlUpdateRequest, do nothing
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, Utils.Infinite_Time,
          ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, operationTime, Utils.Infinite_Time);
      // replicateType is RequestOrResponseType.DeleteRequest, will write the delete record even the source replica doesn't have it.
      cluster.time.sleep(1000);
      operationTime = cluster.time.milliseconds();
      replicateDeleteBlob(targetChannel, blobId, sourceDataNode, operationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
      checkDeleteRecord(targetChannel, clusterMap, blobId, operationTime);
      cluster.time.sleep(1000);

      // 9. On the sourceDatanode blob is ttlUpdated. On the targetDataNode, the blob is deleted
      // ReplicateBlob still succeeds. Nothing is changed on the local store.
      blobId = blobId9;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      long oldOperationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, blobId, oldOperationTime);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      deleteBlob(targetChannel, blobId, oldOperationTime, ServerErrorCode.NoError);
      cluster.time.sleep(1000);
      operationTime = cluster.time.milliseconds();
      // replicateType is RequestOrResponse.PutRequest, do nothing
      replicatePutBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.BadRequest);
      // replicateType is RequestOrResponseType.TtlUpdateRequest, do nothing
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, Utils.Infinite_Time,
          ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
      // replicateType is RequestOrResponseType.DeleteRequest, do nothing
      replicateDeleteBlob(targetChannel, blobId, sourceDataNode, operationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      // opeartion time should be the oldOperationTime
      checkDeleteRecord(targetChannel, clusterMap, blobId, oldOperationTime);
      cluster.time.sleep(1000);

      //
      // souceDataNode has PutBlob, TtlUpdate and DeleteRecord.
      //
      // 10. On the sourceDataNode, the blob is deleted. On the targetDataNode, it doesn't exist.
      // ReplicateBlob replicates PutBlob, TtlUpdate and Delete.
      blobId = blobId10;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      long ttlOperationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, blobId, ttlOperationTime);
      cluster.time.sleep(1000);
      long delOperationTime = cluster.time.milliseconds();
      deleteBlob(sourceChannel, blobId, delOperationTime, ServerErrorCode.NoError);
      // replicateType is RequestOrResponseType.TtlUpdateRequest, will replicate the PutBlob, TtlUpdate and Delete since it doesn't exist
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, Utils.Infinite_Time,
          ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      replicateDeleteBlob(targetChannel, blobId, sourceDataNode, operationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      // The TtlUpdate and Delete operation is same as the source replica's PutBlob creation time.
      // It's not same as the source replica's TtlUpdate or Delete operation time. But it's ok.
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, propertiesWithTtl.getCreationTimeInMs(),
          Utils.Infinite_Time);
      checkDeleteRecord(targetChannel, clusterMap, blobId, propertiesWithTtl.getCreationTimeInMs());
      cluster.time.sleep(1000);

      // 11. On the sourceDataNode, the blob is deleted. TargetDataNode has the PutBlob.
      // ReplicateBlob replicates the deleteBlob
      blobId = blobId11;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      ttlOperationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, blobId, ttlOperationTime);
      cluster.time.sleep(1000);
      delOperationTime = cluster.time.milliseconds();
      deleteBlob(sourceChannel, blobId, delOperationTime, ServerErrorCode.NoError);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      // ReplicatePutBlob do nothing
      replicatePutBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.BadRequest);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      // ReplicateDeleteRecord: it won't have the TtlUpdate Record.
      replicateDeleteBlob(targetChannel, blobId, sourceDataNode, delOperationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkDeleteRecord(targetChannel, clusterMap, blobId,
          delOperationTime); // should be delOperationTime not operationTime
      cluster.time.sleep(1000);

      // 12. On the sourceDataNode, the blob is deleted. On the targetDataNode, it's ttlUpdated.
      // Three ReplicateBlob Types
      blobId = blobId12;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      delOperationTime = cluster.time.milliseconds();
      deleteBlob(sourceChannel, blobId, delOperationTime, ServerErrorCode.NoError);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      cluster.time.sleep(1000);
      ttlOperationTime = cluster.time.milliseconds();
      updateBlobTtl(targetChannel, blobId, ttlOperationTime);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // ReplicatePut do nothing
      replicatePutBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.BadRequest);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // ReplicateTtlUpdate, it already has the TtlUpdate. Do nothing.
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, Utils.Infinite_Time,
          ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // the timestamp should be the ttlOperationTime
      checkTtlUpdateRecord(targetChannel, clusterMap, blobId, ttlOperationTime, Utils.Infinite_Time);
      // ReplicateDelete, it repair the DeleteRecord.
      replicateDeleteBlob(targetChannel, blobId, sourceDataNode, delOperationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkDeleteRecord(targetChannel, clusterMap, blobId, delOperationTime);
      cluster.time.sleep(1000);

      // 13. On the sourceDataNode, the blob is deleted. On the targetDataNode, it's also deleted.
      // ReplicateBlob still succeeds
      blobId = blobId13;
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      delOperationTime = cluster.time.milliseconds();
      deleteBlob(sourceChannel, blobId, delOperationTime, ServerErrorCode.NoError);
      putRequest = new PutRequest(1, "client1", blobId, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      deleteBlob(targetChannel, blobId, delOperationTime, ServerErrorCode.NoError);
      // do nothing for the three repair types
      replicatePutBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.BadRequest);
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, Utils.Infinite_Time,
          ServerErrorCode.NoError);
      replicateDeleteBlob(targetChannel, blobId, sourceDataNode, operationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      // operationTime should be delOperationTime
      checkDeleteRecord(targetChannel, clusterMap, blobId, delOperationTime);
      cluster.time.sleep(1000);

      // 14. On the sourceDataNode, the blob doesn't exist.
      // ReplicateBlob with DeleteOperation is successful
      blobId = blobId14;
      operationTime = cluster.time.milliseconds();
      replicatePutBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.BadRequest);
      replicateTtlUpdate(targetChannel, blobId, sourceDataNode, operationTime, Utils.Infinite_Time,
          ServerErrorCode.BlobNotFound);
      replicateDeleteBlob(targetChannel, blobId, sourceDataNode, operationTime, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      cluster.time.sleep(1000);

      // 15. On the sourceDataNode blob is deleted and compacted. On the targetDataNode, blob doesn't exist.
      BlobId compactedBlobId1 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      // Force delete compactedBlobId1 on source replica. It is used to simulate the compacted tombstone.
      short lifeVersion = 0;
      ForceDeleteAdminRequest forceDeleteAdminRequest = new ForceDeleteAdminRequest(compactedBlobId1, lifeVersion,
          new AdminRequest(AdminRequestOrResponseType.ForceDelete, partitionIds.get(0), 1, "clientid"));
      DataInputStream stream = sourceChannel.sendAndReceive(forceDeleteAdminRequest).getInputStream();
      AdminResponse adminResponse = AdminResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, adminResponse.getError());
      // verify compactedBlobId1 only has a delete tombstone.
      getBlobAndVerify(compactedBlobId1, sourceChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata,
          data, encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
      MessageInfo sourceDelInfo = getDeleteRecord(sourceChannel, clusterMap, compactedBlobId1);
      delOperationTime = sourceDelInfo.getOperationTimeMs();
      // ReplicateBlob only replicate the delete record to the target host.
      replicateDeleteBlob(targetChannel, compactedBlobId1, sourceDataNode, delOperationTime, ServerErrorCode.NoError);
      getBlobAndVerify(compactedBlobId1, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata,
          data, encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
      checkDeleteRecord(targetChannel, clusterMap, compactedBlobId1, delOperationTime);

      // 16. On the sourceDataNode blob is deleted and compacted. On the targetDataNode, blob exists
      BlobId compactedBlobId2 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      // Force delete compactedBlobId2 on the source host
      lifeVersion = 0;
      forceDeleteAdminRequest = new ForceDeleteAdminRequest(compactedBlobId2, lifeVersion,
          new AdminRequest(AdminRequestOrResponseType.ForceDelete, partitionIds.get(0), 1, "clientid2"));
      stream = sourceChannel.sendAndReceive(forceDeleteAdminRequest).getInputStream();
      adminResponse = AdminResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, adminResponse.getError());
      // verify compactedBlobId2 only has a delete tombstone.
      getBlobAndVerify(compactedBlobId2, sourceChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata,
          data, encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
      sourceDelInfo = getDeleteRecord(sourceChannel, clusterMap, compactedBlobId2);
      delOperationTime = sourceDelInfo.getOperationTimeMs();
      // PutBlob on the target
      putRequest = new PutRequest(1, "client1", compactedBlobId2, propertiesWithTtl, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      // ReplicateBlob only replicate the delete record to the target host.
      replicateDeleteBlob(targetChannel, compactedBlobId2, sourceDataNode, delOperationTime, ServerErrorCode.NoError);
      getBlobAndVerify(compactedBlobId2, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
      checkDeleteRecord(targetChannel, clusterMap, compactedBlobId2, delOperationTime);

      /*
       * Stage 2: Enable regular replication.
       * sourceDataNode, targetDataNode and thirdDataNode should be synced up after the regular replication.
                        sourceDataNode targetDataNode   thirdDataNode   targetAfterODR  sourceA/target/thirdAfter
        blobId1              put           not_exist        not_exist         not_exist       put
        blobId2              put           not_exist        put/ttl           put/ttl         ttl
        blobId3              put           not_exist        put/delete        put/delete      delete
        blobId4              put           not_exist        put/delete        put/ttl         delete
        blobId5              put           not_exist        put/ttl           put/delete      delete

        blobId6            put/ttl         not_exist        not_exist         put/ttl         ttl
        blobId7            put/ttl         put              not_exist         put/ttl         ttl
        blobId8            put/ttl         put/ttl          not_exist         put/ttl/del     deleted or not_exist
        blobId9            put/ttl         put/del          not_exist         put/del         deleted or not_exist

        blobId10           put/ttl/del     not_exist        not_exist         put/ttl/del     deleted or not_exist
        blobId11           put/ttl/del     put              not_exist         put/del         deleted or not_exist
        blobId12           put/del         put/ttl          not_exist         put/ttl/del     deleted or not_exist
        blobId13           put/del         put/del          not_exist         put/del         deleted or not_exist

        blobId14           not_exist       not_exist        deleted           not_exist       not_exist
        compactedBlobId1   tombstone       not_exist        not_exist         del             not_exist or del
        compactedBlobId2   tombstone       put              not_exist         put/del         not_exist or del
       */

      // create a new Blob on the sourceDataNode which is TtlUpdated.
      BlobId newBlobId1 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      putRequest = new PutRequest(1, "client1", newBlobId1, properties, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      long newBlobTtlOperationTime = cluster.time.milliseconds();
      updateBlobTtl(sourceChannel, newBlobId1, newBlobTtlOperationTime);
      cluster.time.sleep(1000);

      // create a new blob on the targetDataNode which is deleted.
      BlobId newBlobId2 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      putRequest = new PutRequest(1, "client1", newBlobId2, properties, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      long newBlobDeleteOperationTime = cluster.time.milliseconds();
      deleteBlob(targetChannel, newBlobId2, newBlobDeleteOperationTime, ServerErrorCode.NoError);
      // create the same blob on sourceDataNode as well. So regular replication will replicate the Delete.
      putRequest = new PutRequest(1, "client1", newBlobId2, properties, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      cluster.time.sleep(1000);

      // create a new blob on the thirdDataNode
      BlobId newBlobId3 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      putRequest = new PutRequest(1, "client1", newBlobId3, properties, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, thirdChannel, ServerErrorCode.NoError);
      cluster.time.sleep(1000);

      // Now start the replication on all the replicas
      controlReplicationForPartition(allNodes, partitionId, clientSSLConfig, null, true);

      // wait for the replication to catch up
      notificationSystem.awaitBlobUpdates(newBlobId1.getID(), UpdateType.TTL_UPDATE);
      notificationSystem.awaitBlobDeletions(newBlobId2.getID());
      notificationSystem.awaitBlobCreations(newBlobId3.getID());

      for (int i = 0; i < 3; i++) {
        ConnectedChannel channel;
        if (i == 0) {
          channel = sourceChannel;
        } else if (i == 1) {
          channel = targetChannel;
        } else {
          channel = thirdChannel;
        }

        getBlobAndVerify(blobId1, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);

        getBlobAndVerify(blobId2, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        checkTtlUpdateStatus(channel, clusterMap, blobIdFactory, blobId2, data, true, Utils.Infinite_Time);

        getBlobAndVerify(blobId3, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);

        getBlobAndVerify(blobId4, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);

        getBlobAndVerify(blobId5, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);

        getBlobAndVerify(blobId6, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        checkTtlUpdateStatus(channel, clusterMap, blobIdFactory, blobId6, data, true, Utils.Infinite_Time);

        getBlobAndVerify(blobId7, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        checkTtlUpdateStatus(channel, clusterMap, blobIdFactory, blobId7, data, true, Utils.Infinite_Time);

        if (channel != thirdChannel) {
          getBlobAndVerify(blobId8, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
          getBlobAndVerify(blobId8, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_Deleted_Blobs);
        } else {
          getBlobAndVerify(blobId8, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        }

        if (channel != thirdChannel) {
          getBlobAndVerify(blobId9, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
          getBlobAndVerify(blobId9, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_Deleted_Blobs);
        } else {
          getBlobAndVerify(blobId9, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        }

        if (channel != thirdChannel) {
          getBlobAndVerify(blobId10, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
          getBlobAndVerify(blobId10, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_Deleted_Blobs);
        } else {
          getBlobAndVerify(blobId10, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        }

        if (channel != thirdChannel) {
          getBlobAndVerify(blobId11, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
          getBlobAndVerify(blobId11, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_Deleted_Blobs);
        } else {
          getBlobAndVerify(blobId11, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        }

        if (channel != thirdChannel) {
          getBlobAndVerify(blobId12, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
          getBlobAndVerify(blobId12, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_Deleted_Blobs);
        } else {
          getBlobAndVerify(blobId12, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        }

        if (channel != thirdChannel) {
          getBlobAndVerify(blobId13, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
          getBlobAndVerify(blobId13, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_Deleted_Blobs);
        } else {
          getBlobAndVerify(blobId13, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        }

        if (channel == targetChannel) {
          getBlobAndVerify(blobId14, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        } else {
          getBlobAndVerify(blobId14, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        }

        if (channel == targetChannel || channel == sourceChannel) {
          getBlobAndVerify(compactedBlobId1, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata,
              data, encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_Deleted_Blobs);
        } else {
          getBlobAndVerify(compactedBlobId1, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata,
              data, encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        }

        if (channel == sourceChannel) {
          getBlobAndVerify(compactedBlobId2, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata,
              data, encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_Deleted_Blobs);
        } else if (channel == targetChannel) {
          getBlobAndVerify(compactedBlobId2, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata,
              data, encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
          getBlobAndVerify(compactedBlobId2, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_Deleted_Blobs);
        } else {
          getBlobAndVerify(compactedBlobId2, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata,
              data, encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.Include_All);
        }

        // newBlobId1
        getBlobAndVerify(newBlobId1, channel, ServerErrorCode.NoError, properties, userMetadata, data, encryptionKey,
            clusterMap, blobIdFactory, testEncryption, GetOption.None);
        if (channel == sourceChannel) {
          checkTtlUpdateRecord(channel, clusterMap, newBlobId1, newBlobTtlOperationTime, Utils.Infinite_Time);
        } else {
          // It the local replica doesn't have the PutBlob, when the replication thread replicate the PutBlob,
          // the ttlupdate operation time is same as the creation time.
          checkTtlUpdateRecord(channel, clusterMap, newBlobId1, properties.getCreationTimeInMs(), Utils.Infinite_Time);
        }

        // newBlobId2
        if (channel == targetChannel || channel == sourceChannel) {
          // If the local store has the PutBlob, when we replicate the TtlUpdate in the background thread, we preserve the operationTime.
          getBlobAndVerify(newBlobId2, channel, ServerErrorCode.BlobDeleted, properties, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption, GetOption.None);
          checkDeleteRecord(channel, clusterMap, newBlobId2, newBlobDeleteOperationTime);
        } else {
          // third channel can be NOT_FOUND or Deleted.
        }

        getBlobAndVerify(newBlobId3, channel, ServerErrorCode.NoError, properties, userMetadata, data, encryptionKey,
            clusterMap, blobIdFactory, testEncryption, GetOption.None);
      } // for loop to verify the three DataNode

      sourceChannel.disconnect();
      targetChannel.disconnect();
      thirdChannel.disconnect();
    } catch (Exception e) {
      e.printStackTrace();
      assertNull(e);
    } finally {
      List<? extends ReplicaId> replicaIds = cluster.getClusterMap()
          .getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS)
          .get(0)
          .getReplicaIds();
      for (ReplicaId replicaId : replicaIds) {
        MockReplicaId mockReplicaId = (MockReplicaId) replicaId;
        ((MockDiskId) mockReplicaId.getDiskId()).setDiskState(HardwareState.AVAILABLE, true);
      }
    }
  }
Test ReplicateBlob V2 code under different conditions.
1. test the data correctness of the on-demand replication on the target DataNode.
2. test the interaction between the regular replication and on-demand replication.
ambry-test-utils/src/main/java/com/github/ambry/server/ServerTestUtil.java