static void endToEndReplicationWithMultiNodeMultiPartitionTest(int interestedDataNodePortNumber, Port dataNode1Port,
      Port dataNode2Port, Port dataNode3Port, MockCluster cluster, SSLConfig clientSSLConfig1,
      SSLConfig clientSSLConfig2, SSLConfig clientSSLConfig3, SSLSocketFactory clientSSLSocketFactory1,
      SSLSocketFactory clientSSLSocketFactory2, SSLSocketFactory clientSSLSocketFactory3,
      MockNotificationSystem notificationSystem, boolean testEncryption) throws Exception {
    // interestedDataNodePortNumber is used to locate the datanode and hence has to be PlainTextPort
    MockClusterMap clusterMap = cluster.getClusterMap();
    BlobIdFactory blobIdFactory = new BlobIdFactory(clusterMap);
    List<AmbryServer> serverList = cluster.getServers();
    byte[] usermetadata = new byte[100];
    byte[] data = new byte[100];
    byte[] encryptionKey = null;
    short accountId = Utils.getRandomShort(TestUtils.RANDOM);
    short containerId = Utils.getRandomShort(TestUtils.RANDOM);
    BlobProperties properties =
        new BlobProperties(100, "serviceid1", null, null, false, TestUtils.TTL_SECS, cluster.time.milliseconds(),
            accountId, containerId, false, null, null, null, null);
    long expectedExpiryTimeMs = getExpiryTimeMs(properties);
    TestUtils.RANDOM.nextBytes(usermetadata);
    TestUtils.RANDOM.nextBytes(data);
    if (testEncryption) {
      encryptionKey = new byte[100];
      TestUtils.RANDOM.nextBytes(encryptionKey);
    }

    // connect to all the servers
    ConnectedChannel channel1 =
        getBlockingChannelBasedOnPortType(dataNode1Port, "localhost", clientSSLSocketFactory1, clientSSLConfig1);
    ConnectedChannel channel2 =
        getBlockingChannelBasedOnPortType(dataNode2Port, "localhost", clientSSLSocketFactory2, clientSSLConfig2);
    ConnectedChannel channel3 =
        getBlockingChannelBasedOnPortType(dataNode3Port, "localhost", clientSSLSocketFactory3, clientSSLConfig3);

    // put all the blobs to random servers
    channel1.connect();
    channel2.connect();
    channel3.connect();

    int noOfParallelThreads = 3;
    int totalBlobsToPut = 50;
    CountDownLatch latch = new CountDownLatch(noOfParallelThreads);
    List<DirectSender> runnables = new ArrayList<DirectSender>(noOfParallelThreads);
    ConnectedChannel channel = null;
    for (int i = 0; i < noOfParallelThreads; i++) {
      if (i % noOfParallelThreads == 0) {
        channel = channel1;
      } else if (i % noOfParallelThreads == 1) {
        channel = channel2;
      } else if (i % noOfParallelThreads == 2) {
        channel = channel3;
      }
      DirectSender runnable =
          new DirectSender(cluster, channel, totalBlobsToPut, data, usermetadata, properties, encryptionKey, latch);
      runnables.add(runnable);
      Thread threadToRun = new Thread(runnable);
      threadToRun.start();
    }
    assertTrue("Did not put all blobs in 2 minutes", latch.await(2, TimeUnit.MINUTES));
    // wait till replication can complete
    List<BlobId> blobIds = new ArrayList<BlobId>();
    for (int i = 0; i < runnables.size(); i++) {
      blobIds.addAll(runnables.get(i).getBlobIds());
    }
    for (BlobId blobId : blobIds) {
      notificationSystem.awaitBlobCreations(blobId.getID());
    }

    // Now that the blob is created and replicated, test the cases where a put request arrives for the same blob id
    // later than replication.
    testLatePutRequest(blobIds.get(0), properties, usermetadata, data, encryptionKey, channel1, channel2, channel3,
        ServerErrorCode.NoError);
    // Test the case where a put arrives with the same id as one in the server, but the blob is not identical.
    BlobProperties differentProperties =
        new BlobProperties(properties.getBlobSize(), properties.getServiceId(), accountId, containerId, testEncryption,
            cluster.time.milliseconds());
    testLatePutRequest(blobIds.get(0), differentProperties, usermetadata, data, encryptionKey, channel1, channel2,
        channel3, ServerErrorCode.BlobAlreadyExists);
    byte[] differentUserMetadata = Arrays.copyOf(usermetadata, usermetadata.length);
    differentUserMetadata[0] = (byte) ~differentUserMetadata[0];
    testLatePutRequest(blobIds.get(0), properties, differentUserMetadata, data, encryptionKey, channel1, channel2,
        channel3, ServerErrorCode.BlobAlreadyExists);
    byte[] differentData = Arrays.copyOf(data, data.length);
    differentData[0] = (byte) ~differentData[0];
    testLatePutRequest(blobIds.get(0), properties, usermetadata, differentData, encryptionKey, channel1, channel2,
        channel3, ServerErrorCode.BlobAlreadyExists);

    // verify blob properties, metadata and blob across all nodes
    for (int i = 0; i < 3; i++) {
      channel = null;
      if (i == 0) {
        channel = channel1;
      } else if (i == 1) {
        channel = channel2;
      } else if (i == 2) {
        channel = channel3;
      }

      ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<PartitionRequestInfo>();
      for (int j = 0; j < blobIds.size(); j++) {
        ArrayList<BlobId> ids = new ArrayList<BlobId>();
        ids.add(blobIds.get(j));
        partitionRequestInfoList.clear();
        PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(blobIds.get(j).getPartition(), ids);
        partitionRequestInfoList.add(partitionRequestInfo);
        GetRequest getRequest =
            new GetRequest(1, "clientid2", MessageFormatFlags.BlobProperties, partitionRequestInfoList, GetOption.None);
        DataInputStream stream = channel.sendAndReceive(getRequest).getInputStream();
        GetResponse resp = GetResponse.readFrom(stream, clusterMap);
        try {
          BlobProperties propertyOutput = MessageFormatRecord.deserializeBlobProperties(resp.getInputStream());
          assertEquals(100, propertyOutput.getBlobSize());
          assertEquals("serviceid1", propertyOutput.getServiceId());
          assertEquals("AccountId mismatch", accountId, propertyOutput.getAccountId());
          assertEquals("ContainerId mismatch", containerId, propertyOutput.getContainerId());
          assertEquals("Expiration time mismatch (props)", expectedExpiryTimeMs, getExpiryTimeMs(propertyOutput));
          Assert.assertEquals("Expiration time mismatch (MessageInfo)", expectedExpiryTimeMs,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
          releaseNettyBufUnderneathStream(stream);
        } catch (MessageFormatException e) {
          fail();
        }
        // get user metadata
        getRequest = new GetRequest(1, "clientid2", MessageFormatFlags.BlobUserMetadata, partitionRequestInfoList,
            GetOption.None);
        stream = channel.sendAndReceive(getRequest).getInputStream();
        resp = GetResponse.readFrom(stream, clusterMap);
        try {
          ByteBuffer userMetadataOutput = MessageFormatRecord.deserializeUserMetadata(resp.getInputStream());
          assertArrayEquals(usermetadata, userMetadataOutput.array());
          if (testEncryption) {
            assertNotNull("MessageMetadata should not have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
            Assert.assertArrayEquals("EncryptionKey mismatch", encryptionKey,
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0).getEncryptionKey().array());
          } else {
            assertNull("MessageMetadata should have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
          }
          Assert.assertEquals("Expiration time mismatch (MessageInfo)", expectedExpiryTimeMs,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
        } catch (MessageFormatException e) {
          e.printStackTrace();
          fail();
        }
        releaseNettyBufUnderneathStream(stream);
        // get blob
        getRequest = new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
        stream = channel.sendAndReceive(getRequest).getInputStream();
        resp = GetResponse.readFrom(stream, clusterMap);
        try {
          BlobData blobData = MessageFormatRecord.deserializeBlob(resp.getInputStream());
          byte[] blobout = getBlobData(blobData);
          assertArrayEquals(data, blobout);
          if (testEncryption) {
            assertNotNull("MessageMetadata should not have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
            Assert.assertArrayEquals("EncryptionKey mismatch", encryptionKey,
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0).getEncryptionKey().array());
          } else {
            assertNull("MessageMetadata should have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
          }
          Assert.assertEquals("Expiration time mismatch (MessageInfo)", expectedExpiryTimeMs,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
        } catch (MessageFormatException e) {
          e.printStackTrace();
          fail();
        }
        releaseNettyBufUnderneathStream(stream);
        // get blob all
        getRequest = new GetRequest(1, "clientid2", MessageFormatFlags.All, partitionRequestInfoList, GetOption.None);
        stream = channel.sendAndReceive(getRequest).getInputStream();
        resp = GetResponse.readFrom(stream, clusterMap);
        try {
          BlobAll blobAll = MessageFormatRecord.deserializeBlobAll(resp.getInputStream(), blobIdFactory);
          assertEquals("Expiration time mismatch (props)", expectedExpiryTimeMs,
              getExpiryTimeMs(blobAll.getBlobInfo().getBlobProperties()));
          Assert.assertEquals("Expiration time mismatch (MessageInfo)", expectedExpiryTimeMs,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
          byte[] blobout = getBlobData(blobAll.getBlobData());
          assertArrayEquals(data, blobout);
          if (testEncryption) {
            assertNotNull("EncryptionKey should not ne null", blobAll.getBlobEncryptionKey());
            assertArrayEquals("EncryptionKey mismatch", encryptionKey, blobAll.getBlobEncryptionKey().array());
          } else {
            assertNull("EncryptionKey should have been null", blobAll.getBlobEncryptionKey());
          }
          releaseNettyBufUnderneathStream(stream);
        } catch (MessageFormatException e) {
          e.printStackTrace();
          fail();
        }
      }
    }
    // ttl update all blobs and wait for replication
    Map<ConnectedChannel, List<BlobId>> channelToBlobIds = new HashMap<>();
    for (int i = 0; i < blobIds.size(); i++) {
      final BlobId blobId = blobIds.get(i);
      if (i % 3 == 0) {
        channelToBlobIds.computeIfAbsent(channel1, updateChannel -> new ArrayList<>()).add(blobId);
      } else if (i % 3 == 1) {
        channelToBlobIds.computeIfAbsent(channel2, updateChannel -> new ArrayList<>()).add(blobId);
      } else {
        channelToBlobIds.computeIfAbsent(channel3, updateChannel -> new ArrayList<>()).add(blobId);
      }
    }

    channelToBlobIds.entrySet().stream().map(entry -> CompletableFuture.supplyAsync(() -> {
      try {
        for (BlobId blobId : entry.getValue()) {
          updateBlobTtl(entry.getKey(), blobId, cluster.time.milliseconds());
        }
        return null;
      } catch (Throwable e) {
        throw new RuntimeException("Exception updating ttl for: " + entry, e);
      }
    })).forEach(CompletableFuture::join);

    // check that the TTL update has propagated
    blobIds.forEach(blobId -> notificationSystem.awaitBlobUpdates(blobId.getID(), UpdateType.TTL_UPDATE));
    // check all servers
    for (ConnectedChannel channelToUse : new ConnectedChannel[]{channel1, channel2, channel3}) {
      for (BlobId blobId : blobIds) {
        checkTtlUpdateStatus(channelToUse, clusterMap, blobIdFactory, blobId, data, true, Utils.Infinite_Time);
      }
    }

    // delete random blobs, wait for replication and ensure it is deleted in all nodes

    Set<BlobId> blobsDeleted = new HashSet<BlobId>();
    Set<BlobId> blobsChecked = new HashSet<BlobId>();
    for (int i = 0; i < blobIds.size(); i++) {
      int j = new Random().nextInt(3);
      if (j == 0) {
        j = new Random().nextInt(3);
        if (j == 0) {
          channel = channel1;
        } else if (j == 1) {
          channel = channel2;
        } else if (j == 2) {
          channel = channel3;
        }
        DeleteRequest deleteRequest = new DeleteRequest(1, "reptest", blobIds.get(i), System.currentTimeMillis());
        DataInputStream deleteResponseStream = channel.sendAndReceive(deleteRequest).getInputStream();
        DeleteResponse deleteResponse = DeleteResponse.readFrom(deleteResponseStream);
        releaseNettyBufUnderneathStream(deleteResponseStream);
        Assert.assertEquals(ServerErrorCode.NoError, deleteResponse.getError());
        blobsDeleted.add(blobIds.get(i));
      }
    }

    Iterator<BlobId> iterator = blobsDeleted.iterator();
    ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<PartitionRequestInfo>();
    while (iterator.hasNext()) {
      BlobId deletedId = iterator.next();
      notificationSystem.awaitBlobDeletions(deletedId.getID());
      for (int j = 0; j < 3; j++) {
        if (j == 0) {
          channel = channel1;
        } else if (j == 1) {
          channel = channel2;
        } else if (j == 2) {
          channel = channel3;
        }

        ArrayList<BlobId> ids = new ArrayList<BlobId>();
        ids.add(deletedId);
        partitionRequestInfoList.clear();
        PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(deletedId.getPartition(), 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);
        Assert.assertEquals(ServerErrorCode.BlobDeleted, resp.getPartitionResponseInfoList().get(0).getErrorCode());
        releaseNettyBufUnderneathStream(stream);
      }
    }
    // take a server down, clean up a mount path, start and ensure replication fixes it
    serverList.get(0).shutdown();
    serverList.get(0).awaitShutdown();

    MockDataNodeId dataNode = (MockDataNodeId) clusterMap.getDataNodeId("localhost", interestedDataNodePortNumber);
    System.out.println("Cleaning mount path " + dataNode.getMountPaths().get(0));
    for (ReplicaId replicaId : clusterMap.getReplicaIds(dataNode)) {
      if (replicaId.getMountPath().compareToIgnoreCase(dataNode.getMountPaths().get(0)) == 0) {
        System.out.println("Cleaning partition " + replicaId.getPartitionId());
      }
    }
    deleteFolderContent(new File(dataNode.getMountPaths().get(0)), false);
    for (int i = 0; i < blobIds.size(); i++) {
      for (ReplicaId replicaId : blobIds.get(i).getPartition().getReplicaIds()) {
        if (replicaId.getMountPath().compareToIgnoreCase(dataNode.getMountPaths().get(0)) == 0) {
          if (blobsDeleted.contains(blobIds.get(i))) {
            notificationSystem.decrementDeletedReplica(blobIds.get(i).getID(), dataNode.getHostname(),
                dataNode.getPort());
          } else {
            notificationSystem.decrementCreatedReplica(blobIds.get(i).getID(), dataNode.getHostname(),
                dataNode.getPort());
            notificationSystem.decrementUpdatedReplica(blobIds.get(i).getID(), dataNode.getHostname(),
                dataNode.getPort(), UpdateType.TTL_UPDATE);
          }
        }
      }
    }
    cluster.reinitServer(0);

    channel1.disconnect();
    channel1.connect();

    for (int j = 0; j < blobIds.size(); j++) {
      if (blobsDeleted.contains(blobIds.get(j))) {
        notificationSystem.awaitBlobDeletions(blobIds.get(j).getID());
      } else {
        notificationSystem.awaitBlobCreations(blobIds.get(j).getID());
        notificationSystem.awaitBlobUpdates(blobIds.get(j).getID(), UpdateType.TTL_UPDATE);
      }
      ArrayList<BlobId> ids = new ArrayList<BlobId>();
      ids.add(blobIds.get(j));
      partitionRequestInfoList.clear();
      PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(blobIds.get(j).getPartition(), ids);
      partitionRequestInfoList.add(partitionRequestInfo);
      GetRequest getRequest =
          new GetRequest(1, "clientid2", MessageFormatFlags.BlobProperties, partitionRequestInfoList, GetOption.None);
      DataInputStream stream = channel1.sendAndReceive(getRequest).getInputStream();
      GetResponse resp = GetResponse.readFrom(stream, clusterMap);
      if (resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobDeleted
          || resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobNotFound) {
        assertTrue(blobsDeleted.contains(blobIds.get(j)));
      } else {
        try {
          BlobProperties propertyOutput = MessageFormatRecord.deserializeBlobProperties(resp.getInputStream());
          assertEquals(100, propertyOutput.getBlobSize());
          assertEquals("serviceid1", propertyOutput.getServiceId());
          assertEquals("AccountId mismatch", accountId, propertyOutput.getAccountId());
          assertEquals("ContainerId mismatch", containerId, propertyOutput.getContainerId());
          Assert.assertEquals("Expiration time mismatch in MessageInfo", Utils.Infinite_Time,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
        } catch (MessageFormatException e) {
          fail();
        }
      }
      releaseNettyBufUnderneathStream(stream);
      // get user metadata
      getRequest =
          new GetRequest(1, "clientid2", MessageFormatFlags.BlobUserMetadata, partitionRequestInfoList, GetOption.None);
      stream = channel1.sendAndReceive(getRequest).getInputStream();
      resp = GetResponse.readFrom(stream, clusterMap);
      if (resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobDeleted
          || resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobNotFound) {
        assertTrue(blobsDeleted.contains(blobIds.get(j)));
      } else {
        try {
          ByteBuffer userMetadataOutput = MessageFormatRecord.deserializeUserMetadata(resp.getInputStream());
          assertArrayEquals(usermetadata, userMetadataOutput.array());
          if (testEncryption) {
            assertNotNull("MessageMetadata should not have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
            Assert.assertArrayEquals("EncryptionKey mismatch", encryptionKey,
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0).getEncryptionKey().array());
          } else {
            assertNull("MessageMetadata should have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
          }
          Assert.assertEquals("Expiration time mismatch in MessageInfo", Utils.Infinite_Time,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
        } catch (MessageFormatException e) {
          fail();
        }
      }
      releaseNettyBufUnderneathStream(stream);
      // get blob
      getRequest = new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
      stream = channel1.sendAndReceive(getRequest).getInputStream();
      resp = GetResponse.readFrom(stream, clusterMap);
      if (resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobDeleted
          || resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobNotFound) {
        assertTrue(blobsDeleted.contains(blobIds.get(j)));
      } else {
        try {
          BlobData blobData = MessageFormatRecord.deserializeBlob(resp.getInputStream());
          byte[] blobout = getBlobData(blobData);
          assertArrayEquals(data, blobout);
          if (testEncryption) {
            assertNotNull("MessageMetadata should not have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
            Assert.assertArrayEquals("EncryptionKey mismatch", encryptionKey,
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0).getEncryptionKey().array());
          } else {
            assertNull("MessageMetadata should have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
          }
          Assert.assertEquals("Expiration time mismatch in MessageInfo", Utils.Infinite_Time,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
        } catch (MessageFormatException e) {
          fail();
        }
      }
      releaseNettyBufUnderneathStream(stream);
      // get blob all
      getRequest = new GetRequest(1, "clientid2", MessageFormatFlags.All, partitionRequestInfoList, GetOption.None);
      stream = channel1.sendAndReceive(getRequest).getInputStream();
      resp = GetResponse.readFrom(stream, clusterMap);
      if (resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobDeleted
          || resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobNotFound) {
        assertTrue(blobsDeleted.contains(blobIds.get(j)));
        blobsDeleted.remove(blobIds.get(j));
        blobsChecked.add(blobIds.get(j));
      } else {
        try {
          BlobAll blobAll = MessageFormatRecord.deserializeBlobAll(resp.getInputStream(), blobIdFactory);
          byte[] blobout = getBlobData(blobAll.getBlobData());
          assertArrayEquals(data, blobout);
          if (testEncryption) {
            assertNotNull("EncryptionKey should not ne null", blobAll.getBlobEncryptionKey());
            assertArrayEquals("EncryptionKey mismatch", encryptionKey, blobAll.getBlobEncryptionKey().array());
          } else {
            assertNull("EncryptionKey should have been null", blobAll.getBlobEncryptionKey());
          }
          Assert.assertEquals("Expiration time mismatch in MessageInfo", Utils.Infinite_Time,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
        } catch (MessageFormatException e) {
          fail();
        }
      }
      releaseNettyBufUnderneathStream(stream);
    }
    assertEquals(0, blobsDeleted.size());
    // take a server down, clean all contents, start and ensure replication fixes it
    serverList.get(0).shutdown();
    serverList.get(0).awaitShutdown();
    dataNode = (MockDataNodeId) clusterMap.getDataNodeId("localhost", interestedDataNodePortNumber);
    for (int i = 0; i < dataNode.getMountPaths().size(); i++) {
      System.out.println("Cleaning mount path " + dataNode.getMountPaths().get(i));
      for (ReplicaId replicaId : clusterMap.getReplicaIds(dataNode)) {
        if (replicaId.getMountPath().compareToIgnoreCase(dataNode.getMountPaths().get(i)) == 0) {
          System.out.println("Cleaning partition " + replicaId.getPartitionId());
        }
      }
      deleteFolderContent(new File(dataNode.getMountPaths().get(i)), false);
    }
    for (int i = 0; i < blobIds.size(); i++) {
      if (blobsChecked.contains(blobIds.get(i))) {
        notificationSystem.decrementDeletedReplica(blobIds.get(i).getID(), dataNode.getHostname(), dataNode.getPort());
      } else {
        notificationSystem.decrementCreatedReplica(blobIds.get(i).getID(), dataNode.getHostname(), dataNode.getPort());
      }
    }
    cluster.reinitServer(0);

    channel1.disconnect();
    channel1.connect();

    for (int j = 0; j < blobIds.size(); j++) {
      if (blobsChecked.contains(blobIds.get(j))) {
        notificationSystem.awaitBlobDeletions(blobIds.get(j).getID());
      } else {
        notificationSystem.awaitBlobCreations(blobIds.get(j).getID());
      }
      ArrayList<BlobId> ids = new ArrayList<BlobId>();
      ids.add(blobIds.get(j));
      partitionRequestInfoList.clear();
      PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(blobIds.get(j).getPartition(), ids);
      partitionRequestInfoList.add(partitionRequestInfo);
      GetRequest getRequest =
          new GetRequest(1, "clientid2", MessageFormatFlags.BlobProperties, partitionRequestInfoList, GetOption.None);
      DataInputStream stream = channel1.sendAndReceive(getRequest).getInputStream();
      GetResponse resp = GetResponse.readFrom(stream, clusterMap);
      if (resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobDeleted
          || resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobNotFound) {
        assertTrue(blobsChecked.contains(blobIds.get(j)));
      } else {
        try {
          BlobProperties propertyOutput = MessageFormatRecord.deserializeBlobProperties(resp.getInputStream());
          assertEquals(100, propertyOutput.getBlobSize());
          assertEquals("serviceid1", propertyOutput.getServiceId());
          assertEquals("AccountId mismatch", accountId, propertyOutput.getAccountId());
          assertEquals("ContainerId mismatch", containerId, propertyOutput.getContainerId());
        } catch (MessageFormatException e) {
          fail();
        }
      }
      releaseNettyBufUnderneathStream(stream);
      // get user metadata
      getRequest =
          new GetRequest(1, "clientid2", MessageFormatFlags.BlobUserMetadata, partitionRequestInfoList, GetOption.None);
      stream = channel1.sendAndReceive(getRequest).getInputStream();
      resp = GetResponse.readFrom(stream, clusterMap);
      if (resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobDeleted
          || resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobNotFound) {
        assertTrue(blobsChecked.contains(blobIds.get(j)));
      } else {
        try {
          ByteBuffer userMetadataOutput = MessageFormatRecord.deserializeUserMetadata(resp.getInputStream());
          assertArrayEquals(usermetadata, userMetadataOutput.array());
          if (testEncryption) {
            assertNotNull("MessageMetadata should not have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
            Assert.assertArrayEquals("EncryptionKey mismatch", encryptionKey,
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0).getEncryptionKey().array());
          } else {
            assertNull("MessageMetadata should have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
          }
        } catch (MessageFormatException e) {
          fail();
        }
      }
      releaseNettyBufUnderneathStream(stream);
      // get blob
      getRequest = new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
      stream = channel1.sendAndReceive(getRequest).getInputStream();
      resp = GetResponse.readFrom(stream, clusterMap);
      if (resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobDeleted
          || resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobNotFound) {
        assertTrue(blobsChecked.contains(blobIds.get(j)));
      } else {
        try {
          BlobData blobData = MessageFormatRecord.deserializeBlob(resp.getInputStream());
          byte[] blobout = getBlobData(blobData);
          assertArrayEquals(data, blobout);
          if (testEncryption) {
            assertNotNull("MessageMetadata should not have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
            Assert.assertArrayEquals("EncryptionKey mismatch", encryptionKey,
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0).getEncryptionKey().array());
          } else {
            assertNull("MessageMetadata should have been null",
                resp.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
          }
        } catch (MessageFormatException e) {
          fail();
        }
      }
      releaseNettyBufUnderneathStream(stream);
      // get blob all
      getRequest = new GetRequest(1, "clientid2", MessageFormatFlags.All, partitionRequestInfoList, GetOption.None);
      stream = channel1.sendAndReceive(getRequest).getInputStream();
      resp = GetResponse.readFrom(stream, clusterMap);
      if (resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobDeleted
          || resp.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.BlobNotFound) {
        assertTrue(blobsChecked.contains(blobIds.get(j)));
        blobsChecked.remove(blobIds.get(j));
        blobsDeleted.add(blobIds.get(j));
      } else {
        try {
          BlobAll blobAll = MessageFormatRecord.deserializeBlobAll(resp.getInputStream(), blobIdFactory);
          byte[] blobout = getBlobData(blobAll.getBlobData());
          assertArrayEquals(data, blobout);
          if (testEncryption) {
            assertNotNull("EncryptionKey should not ne null", blobAll.getBlobEncryptionKey());
            assertArrayEquals("EncryptionKey mismatch", encryptionKey, blobAll.getBlobEncryptionKey().array());
          } else {
            assertNull("EncryptionKey should have been null", blobAll.getBlobEncryptionKey());
          }
        } catch (MessageFormatException e) {
          fail();
        }
      }
      releaseNettyBufUnderneathStream(stream);
    }
    assertEquals(0, blobsChecked.size());
    short expectedLifeVersion = 1;
    for (int i = 0; i < 2; i++) {
      expectedLifeVersion += i;
      // First undelete all deleted blobs
      for (BlobId deletedId : blobsDeleted) {
        UndeleteRequest undeleteRequest = new UndeleteRequest(2, "reptest", deletedId, System.currentTimeMillis());
        DataInputStream undeleteResponseStream = channel3.sendAndReceive(undeleteRequest).getInputStream();
        UndeleteResponse undeleteResponse = UndeleteResponse.readFrom(undeleteResponseStream);
        releaseNettyBufUnderneathStream(undeleteResponseStream);
        Assert.assertEquals(ServerErrorCode.NoError, undeleteResponse.getError());
        Assert.assertEquals(expectedLifeVersion, undeleteResponse.getLifeVersion());
      }

      Thread.sleep(5000);
      // Then use get request to get all the data back and make sure the lifeVersion is correct
      for (BlobId id : blobsDeleted) {
        // We don't need to wait for blob undeletes, since one of the hosts has Put Record deleted
        // from disk, so undelete this blob would end up replicating Put Record instead of undelete.
        // notificationSystem.awaitBlobUndeletes(id.toString());
        ArrayList<BlobId> ids = new ArrayList<>();
        ids.add(id);
        partitionRequestInfoList.clear();
        PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(id.getPartition(), ids);
        partitionRequestInfoList.add(partitionRequestInfo);

        // get blob all
        GetRequest getRequest =
            new GetRequest(1, "clientid20", MessageFormatFlags.All, partitionRequestInfoList, GetOption.None);
        DataInputStream stream = channel1.sendAndReceive(getRequest).getInputStream();
        GetResponse resp = GetResponse.readFrom(stream, clusterMap);
        Assert.assertEquals(ServerErrorCode.NoError, resp.getError());
        Assert.assertEquals(1, resp.getPartitionResponseInfoList().size());
        Assert.assertEquals(ServerErrorCode.NoError, resp.getPartitionResponseInfoList().get(0).getErrorCode());
        Assert.assertEquals(1, resp.getPartitionResponseInfoList().get(0).getMessageInfoList().size());
        MessageInfo info = resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0);
        assertEquals(expectedLifeVersion, info.getLifeVersion());
        assertFalse(info.isDeleted());
        try {
          BlobAll blobAll = MessageFormatRecord.deserializeBlobAll(resp.getInputStream(), blobIdFactory);
          byte[] blobout = getBlobData(blobAll.getBlobData());
          assertArrayEquals(data, blobout);
          if (testEncryption) {
            assertNotNull("EncryptionKey should not ne null", blobAll.getBlobEncryptionKey());
            assertArrayEquals("EncryptionKey mismatch", encryptionKey, blobAll.getBlobEncryptionKey().array());
          } else {
            assertNull("EncryptionKey should have been null", blobAll.getBlobEncryptionKey());
          }
          Assert.assertEquals("Expiration time mismatch in MessageInfo", Utils.Infinite_Time,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
          releaseNettyBufUnderneathStream(stream);
        } catch (MessageFormatException e) {
          fail();
        }
      }
      for (BlobId id : blobsDeleted) {
        DeleteRequest deleteRequest = new DeleteRequest(1, "reptest", id, System.currentTimeMillis());
        DataInputStream deleteResponseStream = channel.sendAndReceive(deleteRequest).getInputStream();
        DeleteResponse deleteResponse = DeleteResponse.readFrom(deleteResponseStream);
        releaseNettyBufUnderneathStream(deleteResponseStream);
        Assert.assertEquals(ServerErrorCode.NoError, deleteResponse.getError());
      }

      Thread.sleep(1000);
      for (BlobId id : blobsDeleted) {
        ArrayList<BlobId> ids = new ArrayList<>();
        ids.add(id);
        partitionRequestInfoList.clear();
        PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(id.getPartition(), ids);
        partitionRequestInfoList.add(partitionRequestInfo);

        // get blob all
        GetRequest getRequest =
            new GetRequest(1, "clientid200", MessageFormatFlags.All, partitionRequestInfoList, GetOption.Include_All);
        DataInputStream stream = channel1.sendAndReceive(getRequest).getInputStream();
        GetResponse resp = GetResponse.readFrom(stream, clusterMap);
        Assert.assertEquals(ServerErrorCode.NoError, resp.getError());
        Assert.assertEquals(1, resp.getPartitionResponseInfoList().size());
        Assert.assertEquals(ServerErrorCode.NoError, resp.getPartitionResponseInfoList().get(0).getErrorCode());
        Assert.assertEquals(1, resp.getPartitionResponseInfoList().get(0).getMessageInfoList().size());
        MessageInfo info = resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0);
        assertEquals(expectedLifeVersion, info.getLifeVersion());
        assertTrue(info.isDeleted());
        try {
          BlobAll blobAll = MessageFormatRecord.deserializeBlobAll(resp.getInputStream(), blobIdFactory);
          byte[] blobout = getBlobData(blobAll.getBlobData());
          assertArrayEquals(data, blobout);
          if (testEncryption) {
            assertNotNull("EncryptionKey should not ne null", blobAll.getBlobEncryptionKey());
            assertArrayEquals("EncryptionKey mismatch", encryptionKey, blobAll.getBlobEncryptionKey().array());
          } else {
            assertNull("EncryptionKey should have been null", blobAll.getBlobEncryptionKey());
          }
          Assert.assertEquals("Expiration time mismatch in MessageInfo", Utils.Infinite_Time,
              resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs());
          releaseNettyBufUnderneathStream(stream);
        } catch (MessageFormatException e) {
          fail();
        }
      }
    }
    channel1.disconnect();
    channel2.disconnect();
    channel3.disconnect();
  }
Tests the case where a late PutRequest is sent to each server when it already has a record for the given BlobId.
The expected error from each server should be the given error.
@param blobId the {@link BlobId} of the blob to be put.
@param properties the {@link BlobProperties} of the blob to be put.
@param usermetadata the usermetadata of the blob to be put.
@param data the blob data of the blob to be put.
@param encryptionKey the encryption key of the blob. Could be null for non-encrypted blobs
@param channelToDatanode1 the {@link ConnectedChannel} to the Datanode1.
@param channelToDatanode2 the {@link ConnectedChannel} to the Datanode2.
@param channelToDatanode3 the {@link ConnectedChannel} to the Datanode3.
@param expectedErrorCode the {@link ServerErrorCode} that is expected from every Datanode.
@throws IOException
ambry-test-utils/src/main/java/com/github/ambry/server/ServerTestUtil.java