static void replicateBlobCaseTest(MockCluster cluster, SSLConfig clientSSLConfig, boolean testEncryption,
      MockNotificationSystem notificationSystem, boolean writeRepair) {
    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.
     * comparisonDataNode uses the regular replication thread to replicate the Blobs from peer replicas.
     */
    DataNodeId sourceDataNode = dataNodes.get(0);
    DataNodeId targetDataNode = dataNodes.get(1);
    DataNodeId comparisonDataNode = 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 expired blob
      BlobProperties propertiesExpired =
          new BlobProperties(dataSize, "serviceid1", "ownerid", "jpeg", false, 0, cluster.time.milliseconds(),
              accountId, containerId, testEncryption, null, null, null, null);
      // property of blob with the infinite TTL
      BlobProperties properties = new BlobProperties(dataSize, "serviceid1", accountId, containerId, testEncryption,
          cluster.time.milliseconds());

      short blobIdVersion = CommonTestUtils.getCurrentBlobIdVersion();
      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);
      BlobId blobId15 = 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 comparisonPort = new Port(comparisonDataNode.getHttp2Port(), PortType.HTTP2);
      ConnectedChannel sourceChannel =
          getBlockingChannelBasedOnPortType(sourcePort, "localhost", null, clientSSLConfig);
      ConnectedChannel targetChannel =
          getBlockingChannelBasedOnPortType(targetPort, "localhost", null, clientSSLConfig);
      ConnectedChannel comparisonChannel =
          getBlockingChannelBasedOnPortType(comparisonPort, "localhost", null, clientSSLConfig);
      sourceChannel.connect();
      targetChannel.connect();
      comparisonChannel.connect();

      /*
       * 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.
       */
      // 1. On the sourceDataNode, putBlob. On the targetDataNode, blob doesn't exist.
      // ReplicateBlob replicates the blob to the target.
      BlobId blobId = blobId1;
      PutRequest 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);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);

      // 2. On the sourceDataNode, putBlob. On the targetDataNode, the blob already exists.
      // ReplicateBlob to both the sourceDataNode and the targetDataNode. All succeed.
      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);
      replicateBlob(sourceChannel, blobId, sourceDataNode, 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);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);

      // 3, On the sourceDataNode, putBlob. On the targetDataNode, the blob is ttlUpdated.
      // ReplicateBlob still succeeds
      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, targetChannel, ServerErrorCode.NoError);
      updateBlobTtl(targetChannel, blobId, cluster.time.milliseconds());
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);

      // 4. On the sourceDataNode, putBlob. On the targetDataNode, the blob is deleted.
      // ReplicateBlob still succeeds
      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, targetChannel, ServerErrorCode.NoError);
      deleteBlob(targetChannel, blobId, cluster.time.milliseconds(), ServerErrorCode.NoError);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);

      // 5. On the sourceDatanode, the blob is ttlUpdated. On the targetDataNode, it doesn't exist.
      // ReplicateBlob will replicate both the putBlob and the TtlUpdate
      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);
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, false, ttlUpdateBlobExpiryTimeMs);
      updateBlobTtl(sourceChannel, blobId, cluster.time.milliseconds());
      checkTtlUpdateStatus(sourceChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      // replicate to the targetDataNode
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);

      // 6. On the sourceDatanode blob is ttlUpdated. On the targetDataNode, putBlob.
      // ReplicateBlob replicates 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);
      updateBlobTtl(sourceChannel, blobId, cluster.time.milliseconds());
      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);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      if (writeRepair) {
        checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      } else {
        checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, false, ttlUpdateBlobExpiryTimeMs);
      }

      // 7. On the sourceDatanode blob is ttlUpdated. On the targetDataNode, the blob is also ttlUpdated already.
      // ReplicateBlob still succeeds.
      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);
      updateBlobTtl(sourceChannel, blobId, cluster.time.milliseconds());
      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, cluster.time.milliseconds());
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);

      // 8. On the sourceDatanode blob is ttlUpdated. On the targetDataNode, the blob is deleted
      // ReplicateBlob still succeeds. Nothing is changed on the local store.
      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);
      updateBlobTtl(sourceChannel, blobId, cluster.time.milliseconds());
      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, cluster.time.milliseconds(), ServerErrorCode.NoError);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);

      // 9. On the sourceDataNode, the blob is deleted. On the targetDataNode, it doesn't exist.
      // ReplicateBlob replicates both the putBlob and the deleteBlob.
      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);
      deleteBlob(sourceChannel, blobId, cluster.time.milliseconds(), ServerErrorCode.NoError);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);

      // 10. On the sourceDataNode, the blob is deleted. On the targetDataNode, it exists.
      // ReplicateBlob replicates the deleteBlob
      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);
      deleteBlob(sourceChannel, blobId, cluster.time.milliseconds(), 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);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      if (writeRepair) {
        getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
      } else {
        getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
      }

      // 11. On the sourceDataNode, the blob is deleted. On the targetDataNode, it's ttlUpdated.
      // 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);
      deleteBlob(sourceChannel, blobId, cluster.time.milliseconds(), 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);
      updateBlobTtl(targetChannel, blobId, cluster.time.milliseconds());
      checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      if (writeRepair) {
        getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
      } else {
        getBlobAndVerify(blobId, targetChannel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        checkTtlUpdateStatus(targetChannel, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      }

      // 12. On the sourceDataNode, the blob is deleted. On the targetDataNode, it's also deleted.
      // ReplicateBlob still succeeds
      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);
      deleteBlob(sourceChannel, blobId, cluster.time.milliseconds(), 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, cluster.time.milliseconds(), ServerErrorCode.NoError);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, properties, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);

      // 13. On the sourceDataNode, the blob is expired. On the targetDataNode, it doesn't exist.
      // ReplicateBlob replicate the Blob even it's expired.
      blobId = blobId13;
      putRequest = new PutRequest(1, "client1", blobId, propertiesExpired, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      cluster.time.sleep(10000);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobExpired, propertiesExpired, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);

      // 14. On the sourceDataNode, the blob is expired. On the targetDataNode, it's deleted.
      // ReplicateBlob succeeds but nothing is changed on the local store.
      blobId = blobId14;
      putRequest = new PutRequest(1, "client1", blobId, propertiesExpired, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      cluster.time.sleep(10000);
      putRequest = new PutRequest(1, "client1", blobId, propertiesExpired, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);
      deleteBlob(targetChannel, blobId, cluster.time.milliseconds(), ServerErrorCode.NoError);
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.NoError);
      getBlobAndVerify(blobId, targetChannel, ServerErrorCode.BlobDeleted, propertiesExpired, userMetadata, data,
          encryptionKey, clusterMap, blobIdFactory, testEncryption);

      // 15. On the sourceDataNode, the blob doesn't exist.
      // ReplicateBlob fails with Blob_Not_Found status
      blobId = blobId15;
      replicateBlob(targetChannel, blobId, sourceDataNode, ServerErrorCode.BlobNotFound);

      /*
       * Stage 2: Update the blobs on either the sourceDataNode or the targetDataNode.
       * Enable regular replication. sourceDataNode and targetDataNode should be synced up after the regular replication.
       * Other replicas should catch up as well.
               sourceDataNode  targetDataNode  target_after_ReplicateBlob     extra_operation        after_replication
                                               write_repair/non_write_repair
        blob1     putBlob         not_exist       putBlob                   ttlUpdate_on_source       ttlUpdated
        blob2     putBlob         putBlob         putBlob                   ttlUpdate_on_target       ttlUpdated
        blob3     putBlob         ttlUpdated      ttlUpdated                deleteBlob_on_source      deleted, can be not_found
        blob4     putBlob         deleted         deleted                                             deleted, can be not_found
        blob5     ttlUpdated      not_exist       ttlUpdated                                          ttlUpdated
        blob6     ttlUpdated      putBlob         ttlUpdated/putBlob        deleteBlob_on_target      deleted, can be not_found
        blob7     ttlUpdated      ttlUpdated      ttlUpdated                                          ttlUpdated
        blob8     ttlUpdated      deleted         deleted                                             deleted, can be not_found
        blob9     deleted         not_exist       deleted                   undeleteBlob_on_target    undeleted
        blob10    deleted         putBlob         deleted/putBlob           undeleteBlob_on_source    undeleted
        blob11    deleted         ttlUpdated      deleted/ttlUpdated                                  deleted/not_found
        blob12    deleted         deleted         deleted                                             deleted/not_found
        blob13    expired         not_exist       expired                                             expired, can be not_found
        blob14    expired         deleted         deleted                                             deleted, can be expired/not_found
        blob15    not_exist       not_exist       error: Blob_Not_Found                               Not_Found
        blob16                                                              putBlob_on_source         putBlob
        blob17                                                              purBlob_on_target         putBob
       */

      // TtlUpdate blob1 on the sourceDataNode
      updateBlobTtl(sourceChannel, blobId1, cluster.time.milliseconds());
      // TtlUpdate blob2 on the targetDataNode
      updateBlobTtl(targetChannel, blobId2, cluster.time.milliseconds());
      // Delete blob3 on the sourceDataNode
      deleteBlob(sourceChannel, blobId3, cluster.time.milliseconds(), ServerErrorCode.NoError);
      // Delete blob6 on the targetDataNode
      deleteBlob(targetChannel, blobId6, cluster.time.milliseconds(), ServerErrorCode.NoError);
      // Undelete blob9 on the targetDataNode
      undeleteBlob(targetChannel, blobId9, cluster.time.milliseconds(), (short) 1);
      // Undelete blob10 on the sourceDataNode
      undeleteBlob(sourceChannel, blobId10, cluster.time.milliseconds(), (short) 1);
      // create blob16 on the sourceDataNode
      BlobId blobId16 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      putRequest = new PutRequest(1, "client1", blobId16, properties, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, sourceChannel, ServerErrorCode.NoError);
      // create blob17 on the targetDataNode
      BlobId blobId17 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionId, testEncryption,
          BlobId.BlobDataType.DATACHUNK);
      putRequest = new PutRequest(1, "client1", blobId17, properties, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob,
          testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putBlob(putRequest, targetChannel, ServerErrorCode.NoError);

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

      // wait for the replication to catch up
      notificationSystem.awaitBlobCreations(blobId16.getID());
      notificationSystem.awaitBlobCreations(blobId17.getID());

      // verify the status. sourceDataNode and targetDataNode are synced up.
      // the comparisonDataNode replicated the blobs from the peers.
      for (int i = 0; i < 3; i++) {
        ConnectedChannel channel;
        if (i == 0) {
          channel = sourceChannel;
        } else if (i == 1) {
          channel = targetChannel;
        } else {
          channel = comparisonChannel;
        }
        // blob1: TtlUpdate with Infinite
        checkTtlUpdateStatus(channel, clusterMap, blobIdFactory, blobId1, data, true, Utils.Infinite_Time);
        getBlobAndVerify(blobId1, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        // blob2: TtlUpdate with Infinite
        checkTtlUpdateStatus(channel, clusterMap, blobIdFactory, blobId2, data, true, Utils.Infinite_Time);
        getBlobAndVerify(blobId2, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        // blob3: deleted. For the comparisonDataNode, regular replication won't replicate deleted blob. It can be either NOT_FOUND or DELETED.
        if (channel == comparisonChannel) {
          ArrayList<BlobId> ids = new ArrayList<>();
          ids.add(blobId3);
          ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
          PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(partitionId, ids);
          partitionRequestInfoList.add(partitionRequestInfo);
          GetRequest getRequest =
              new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
          DataInputStream stream = channel.sendAndReceive(getRequest).getInputStream();
          GetResponse resp = GetResponse.readFrom(stream, clusterMap);
          ServerErrorCode error = resp.getPartitionResponseInfoList().get(0).getErrorCode();
          assertTrue(error == ServerErrorCode.BlobDeleted || error == ServerErrorCode.BlobNotFound);
          releaseNettyBufUnderneathStream(stream);
        } else {
          getBlobAndVerify(blobId3, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption);
        }
        // blob4: deleted. ComparisonChannel can be NOT_FOUND
        if (channel == comparisonChannel) {
          ArrayList<BlobId> ids = new ArrayList<>();
          ids.add(blobId4);
          ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
          PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(partitionId, ids);
          partitionRequestInfoList.add(partitionRequestInfo);
          GetRequest getRequest =
              new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
          DataInputStream stream = channel.sendAndReceive(getRequest).getInputStream();
          GetResponse resp = GetResponse.readFrom(stream, clusterMap);
          ServerErrorCode error = resp.getPartitionResponseInfoList().get(0).getErrorCode();
          assertTrue(error == ServerErrorCode.BlobDeleted || error == ServerErrorCode.BlobNotFound);
          releaseNettyBufUnderneathStream(stream);
        } else {
          getBlobAndVerify(blobId4, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption);
        }
        // blob5: TtlUpdate with Infinite
        checkTtlUpdateStatus(channel, clusterMap, blobIdFactory, blobId5, data, true, Utils.Infinite_Time);
        getBlobAndVerify(blobId5, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        // blob6: deleted. For the comparisonDataNode, regular replication won't replicate deleted blob. It can be either NOT_FOUND or DELETED.
        if (channel == comparisonChannel) {
          ArrayList<BlobId> ids = new ArrayList<>();
          ids.add(blobId6);
          ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
          PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(partitionId, ids);
          partitionRequestInfoList.add(partitionRequestInfo);
          GetRequest getRequest =
              new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
          DataInputStream stream = channel.sendAndReceive(getRequest).getInputStream();
          GetResponse resp = GetResponse.readFrom(stream, clusterMap);
          ServerErrorCode error = resp.getPartitionResponseInfoList().get(0).getErrorCode();
          assertTrue(error == ServerErrorCode.BlobDeleted || error == ServerErrorCode.BlobNotFound);
          releaseNettyBufUnderneathStream(stream);
        } else {
          getBlobAndVerify(blobId6, channel, ServerErrorCode.BlobDeleted, propertiesExpired, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption);
        }
        // blob7: TtlUpdate with Infinite
        checkTtlUpdateStatus(channel, clusterMap, blobIdFactory, blobId7, data, true, Utils.Infinite_Time);
        getBlobAndVerify(blobId7, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        // blob8: deleted.
        if (channel == comparisonChannel) {
          ArrayList<BlobId> ids = new ArrayList<>();
          ids.add(blobId8);
          ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
          PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(partitionId, ids);
          partitionRequestInfoList.add(partitionRequestInfo);
          GetRequest getRequest =
              new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
          DataInputStream stream = channel.sendAndReceive(getRequest).getInputStream();
          GetResponse resp = GetResponse.readFrom(stream, clusterMap);
          ServerErrorCode error = resp.getPartitionResponseInfoList().get(0).getErrorCode();
          assertTrue(error == ServerErrorCode.BlobDeleted || error == ServerErrorCode.BlobNotFound);
          releaseNettyBufUnderneathStream(stream);
        } else {
          getBlobAndVerify(blobId8, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption);
        }
        // blob9: undeleted
        getBlobAndVerify(blobId9, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        // blob10: undeleted
        getBlobAndVerify(blobId10, channel, ServerErrorCode.NoError, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        // blob11: deleted or not_found
        if (channel == comparisonChannel) {
          if (writeRepair) {
            getBlobAndVerify(blobId11, channel, ServerErrorCode.BlobNotFound, propertiesExpired, userMetadata, data,
                encryptionKey, clusterMap, blobIdFactory, testEncryption);
          } else {
            ArrayList<BlobId> ids = new ArrayList<>();
            ids.add(blobId11);
            ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
            PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(partitionId, ids);
            partitionRequestInfoList.add(partitionRequestInfo);
            GetRequest getRequest =
                new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
            DataInputStream stream = channel.sendAndReceive(getRequest).getInputStream();
            GetResponse resp = GetResponse.readFrom(stream, clusterMap);
            ServerErrorCode error = resp.getPartitionResponseInfoList().get(0).getErrorCode();
            assertTrue(error == ServerErrorCode.BlobDeleted || error == ServerErrorCode.BlobNotFound);
            releaseNettyBufUnderneathStream(stream);
          }
        } else {
          getBlobAndVerify(blobId11, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption);
        }
        // blob12: deleted
        if (channel == comparisonChannel) {
          getBlobAndVerify(blobId12, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption);
        } else {
          getBlobAndVerify(blobId12, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption);
        }
        // blob13: expired
        if (channel == comparisonChannel) {
          ArrayList<BlobId> ids = new ArrayList<>();
          ids.add(blobId13);
          ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
          PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(partitionId, ids);
          partitionRequestInfoList.add(partitionRequestInfo);
          GetRequest getRequest =
              new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
          DataInputStream stream = channel.sendAndReceive(getRequest).getInputStream();
          GetResponse resp = GetResponse.readFrom(stream, clusterMap);
          ServerErrorCode error = resp.getPartitionResponseInfoList().get(0).getErrorCode();
          assertTrue(error == ServerErrorCode.BlobExpired || error == ServerErrorCode.BlobNotFound);
          releaseNettyBufUnderneathStream(stream);
        } else {
          getBlobAndVerify(blobId13, channel, ServerErrorCode.BlobExpired, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption);
        }
        // blob14: deleted
        if (channel == comparisonChannel) {
          ArrayList<BlobId> ids = new ArrayList<>();
          ids.add(blobId14);
          ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
          PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(partitionId, ids);
          partitionRequestInfoList.add(partitionRequestInfo);
          GetRequest getRequest =
              new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
          DataInputStream stream = channel.sendAndReceive(getRequest).getInputStream();
          GetResponse resp = GetResponse.readFrom(stream, clusterMap);
          ServerErrorCode error = resp.getPartitionResponseInfoList().get(0).getErrorCode();
          assertTrue(error == ServerErrorCode.BlobDeleted || error == ServerErrorCode.BlobExpired
              || error == ServerErrorCode.BlobNotFound);
          releaseNettyBufUnderneathStream(stream);
        } else {
          getBlobAndVerify(blobId14, channel, ServerErrorCode.BlobDeleted, propertiesWithTtl, userMetadata, data,
              encryptionKey, clusterMap, blobIdFactory, testEncryption);
        }
        // blob15: not found
        getBlobAndVerify(blobId15, channel, ServerErrorCode.BlobNotFound, propertiesWithTtl, userMetadata, data,
            encryptionKey, clusterMap, blobIdFactory, testEncryption);
        // blob16: newly created with infinite ttl
        getBlobAndVerify(blobId16, channel, ServerErrorCode.NoError, properties, userMetadata, data, encryptionKey,
            clusterMap, blobIdFactory, testEncryption);
        // blob17: newly created with infinite ttl
        getBlobAndVerify(blobId17, channel, ServerErrorCode.NoError, properties, userMetadata, data, encryptionKey,
            clusterMap, blobIdFactory, testEncryption);

        channel.disconnect();
      } // for loop to verify the three DataNode
    } 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 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.
If writeRepair is true, even the local store has the BlobID, ReplicateBlob is still executed and will repair the final state.
If writeRepair is false, if the local store has the BlobID, ReplicateBlob does nothing and returns success status.
ambry-test-utils/src/main/java/com/github/ambry/server/ServerTestUtil.java