static void endToEndReplicationWithMultiNodeSinglePartitionTest(String routerDatacenter,
      int interestedDataNodePortNumber, Port dataNode1Port, Port dataNode2Port, Port dataNode3Port, MockCluster cluster,
      SSLConfig clientSSLConfig1, SSLSocketFactory clientSSLSocketFactory1, MockNotificationSystem notificationSystem,
      Properties routerProps, boolean testEncryption) {
    // interestedDataNodePortNumber is used to locate the datanode and hence has to be PlainText port
    try {
      // The header size of a LogSegment. This shouldn't be here since it breaks the interface of Log. But to satisfy the test cases
      // we will use this number here.
      // This also means we only have one log segment for this partition. If we put more operations to the partition and it excceeds
      // the log segment capacity, this number will have to be increased.
      int expectedTokenSize = 18;
      MockClusterMap clusterMap = cluster.getClusterMap();
      BlobIdFactory blobIdFactory = new BlobIdFactory(clusterMap);
      ArrayList<BlobProperties> propertyList = new ArrayList<>();
      ArrayList<BlobId> blobIdList = new ArrayList<>();
      ArrayList<byte[]> dataList = new ArrayList<>();
      ArrayList<byte[]> encryptionKeyList = new ArrayList<>();
      byte[] usermetadata = new byte[1000];
      TestUtils.RANDOM.nextBytes(usermetadata);
      PartitionId partition = clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS).get(0);

      for (int i = 0; i < 11; i++) {
        short accountId = Utils.getRandomShort(TestUtils.RANDOM);
        short containerId = Utils.getRandomShort(TestUtils.RANDOM);
        propertyList.add(
            new BlobProperties(1000, "serviceid1", null, null, false, TestUtils.TTL_SECS, cluster.time.milliseconds(),
                accountId, containerId, testEncryption, null, null, null, null));
        blobIdList.add(new BlobId(CommonTestUtils.getCurrentBlobIdVersion(), BlobId.BlobIdType.NATIVE,
            clusterMap.getLocalDatacenterId(), accountId, containerId, partition, false,
            BlobId.BlobDataType.DATACHUNK));
        dataList.add(TestUtils.getRandomBytes(1000));
        if (testEncryption) {
          encryptionKeyList.add(TestUtils.getRandomBytes(128));
        } else {
          encryptionKeyList.add(null);
        }
      }

      // put blob 1
      PutRequest putRequest =
          new PutRequest(1, "client1", blobIdList.get(0), propertyList.get(0), ByteBuffer.wrap(usermetadata),
              Unpooled.wrappedBuffer(dataList.get(0)), propertyList.get(0).getBlobSize(), BlobType.DataBlob,
              encryptionKeyList.get(0) != null ? ByteBuffer.wrap(encryptionKeyList.get(0)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(0), blobIdList.get(0),
          encryptionKeyList.get(0) != null ? ByteBuffer.wrap(encryptionKeyList.get(0)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(0));

      ConnectedChannel channel1 =
          getBlockingChannelBasedOnPortType(dataNode1Port, "localhost", clientSSLSocketFactory1, clientSSLConfig1);
      ConnectedChannel channel2 =
          getBlockingChannelBasedOnPortType(dataNode2Port, "localhost", clientSSLSocketFactory1, clientSSLConfig1);
      ConnectedChannel channel3 =
          getBlockingChannelBasedOnPortType(dataNode3Port, "localhost", clientSSLSocketFactory1, clientSSLConfig1);

      channel1.connect();
      channel2.connect();
      channel3.connect();
      DataInputStream putResponseStream = channel1.sendAndReceive(putRequest).getInputStream();
      PutResponse response = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response.getError());
      // put blob 2
      PutRequest putRequest2 =
          new PutRequest(1, "client1", blobIdList.get(1), propertyList.get(1), ByteBuffer.wrap(usermetadata),
              Unpooled.wrappedBuffer(dataList.get(1)), propertyList.get(1).getBlobSize(), BlobType.DataBlob,
              encryptionKeyList.get(1) != null ? ByteBuffer.wrap(encryptionKeyList.get(1)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(1), blobIdList.get(1),
          encryptionKeyList.get(1) != null ? ByteBuffer.wrap(encryptionKeyList.get(1)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(1));
      putResponseStream = channel2.sendAndReceive(putRequest2).getInputStream();
      PutResponse response2 = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response2.getError());
      // put blob 3
      PutRequest putRequest3 =
          new PutRequest(1, "client1", blobIdList.get(2), propertyList.get(2), ByteBuffer.wrap(usermetadata),
              Unpooled.wrappedBuffer(dataList.get(2)), propertyList.get(2).getBlobSize(), BlobType.DataBlob,
              encryptionKeyList.get(2) != null ? ByteBuffer.wrap(encryptionKeyList.get(2)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(2), blobIdList.get(2),
          encryptionKeyList.get(2) != null ? ByteBuffer.wrap(encryptionKeyList.get(2)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(2));
      putResponseStream = channel3.sendAndReceive(putRequest3).getInputStream();
      PutResponse response3 = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response3.getError());

      // put blob 4
      putRequest = new PutRequest(1, "client1", blobIdList.get(3), propertyList.get(3), ByteBuffer.wrap(usermetadata),
          Unpooled.wrappedBuffer(dataList.get(3)), propertyList.get(3).getBlobSize(), BlobType.DataBlob,
          encryptionKeyList.get(3) != null ? ByteBuffer.wrap(encryptionKeyList.get(3)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(3), blobIdList.get(3),
          encryptionKeyList.get(3) != null ? ByteBuffer.wrap(encryptionKeyList.get(3)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(3));
      putResponseStream = channel1.sendAndReceive(putRequest).getInputStream();
      response = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response.getError());

      // put blob 5
      putRequest2 = new PutRequest(1, "client1", blobIdList.get(4), propertyList.get(4), ByteBuffer.wrap(usermetadata),
          Unpooled.wrappedBuffer(dataList.get(4)), propertyList.get(4).getBlobSize(), BlobType.DataBlob,
          encryptionKeyList.get(4) != null ? ByteBuffer.wrap(encryptionKeyList.get(4)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(4), blobIdList.get(4),
          encryptionKeyList.get(4) != null ? ByteBuffer.wrap(encryptionKeyList.get(4)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(4));
      putResponseStream = channel2.sendAndReceive(putRequest2).getInputStream();
      response2 = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response2.getError());

      // put blob 6
      putRequest3 = new PutRequest(1, "client1", blobIdList.get(5), propertyList.get(5), ByteBuffer.wrap(usermetadata),
          Unpooled.wrappedBuffer(dataList.get(5)), propertyList.get(5).getBlobSize(), BlobType.DataBlob,
          encryptionKeyList.get(5) != null ? ByteBuffer.wrap(encryptionKeyList.get(5)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(5), blobIdList.get(5),
          encryptionKeyList.get(5) != null ? ByteBuffer.wrap(encryptionKeyList.get(5)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(5));
      putResponseStream = channel3.sendAndReceive(putRequest3).getInputStream();
      response3 = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response3.getError());
      // wait till replication can complete
      notificationSystem.awaitBlobCreations(blobIdList.get(0).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(1).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(2).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(3).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(4).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(5).getID());

      checkTtlUpdateStatus(channel3, clusterMap, blobIdFactory, blobIdList.get(5), dataList.get(5), false,
          getExpiryTimeMs(propertyList.get(5)));
      updateBlobTtl(channel3, blobIdList.get(5), cluster.time.milliseconds());
      expectedTokenSize += getUpdateRecordSize(blobIdList.get(5), SubRecord.Type.TTL_UPDATE);
      checkTtlUpdateStatus(channel3, clusterMap, blobIdFactory, blobIdList.get(5), dataList.get(5), true,
          Utils.Infinite_Time);
      notificationSystem.awaitBlobUpdates(blobIdList.get(5).getID(), UpdateType.TTL_UPDATE);

      // get blob properties
      ArrayList<BlobId> ids = new ArrayList<BlobId>();
      MockPartitionId mockPartitionId =
          (MockPartitionId) clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS).get(0);
      ids.add(blobIdList.get(2));
      ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<PartitionRequestInfo>();
      PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(mockPartitionId, ids);
      partitionRequestInfoList.add(partitionRequestInfo);
      GetRequest getRequest1 =
          new GetRequest(1, "clientid2", MessageFormatFlags.BlobProperties, partitionRequestInfoList, GetOption.None);
      DataInputStream stream = channel2.sendAndReceive(getRequest1).getInputStream();
      GetResponse resp1 = GetResponse.readFrom(stream, clusterMap);
      Assert.assertEquals(ServerErrorCode.NoError, resp1.getError());
      Assert.assertEquals(ServerErrorCode.NoError, resp1.getPartitionResponseInfoList().get(0).getErrorCode());
      try {
        BlobProperties propertyOutput = MessageFormatRecord.deserializeBlobProperties(resp1.getInputStream());
        assertEquals(1000, propertyOutput.getBlobSize());
        assertEquals("serviceid1", propertyOutput.getServiceId());
        assertEquals("AccountId mismatch", propertyList.get(2).getAccountId(), propertyOutput.getAccountId());
        assertEquals("ContainerId mismatch", propertyList.get(2).getContainerId(), propertyOutput.getContainerId());
        assertEquals("IsEncrypted mismatch", propertyList.get(2).isEncrypted(), propertyOutput.isEncrypted());
      } catch (MessageFormatException e) {
        fail();
      }
      // get user metadata
      ids.clear();
      ids.add(blobIdList.get(1));
      GetRequest getRequest2 =
          new GetRequest(1, "clientid2", MessageFormatFlags.BlobUserMetadata, partitionRequestInfoList, GetOption.None);
      stream = channel1.sendAndReceive(getRequest2).getInputStream();
      GetResponse resp2 = GetResponse.readFrom(stream, clusterMap);
      Assert.assertEquals(ServerErrorCode.NoError, resp2.getError());
      Assert.assertEquals(ServerErrorCode.NoError, resp2.getPartitionResponseInfoList().get(0).getErrorCode());
      try {
        ByteBuffer userMetadataOutput = MessageFormatRecord.deserializeUserMetadata(resp2.getInputStream());
        assertArrayEquals(usermetadata, userMetadataOutput.array());
        if (testEncryption) {
          assertNotNull("MessageMetadata should not have been null",
              resp2.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
          Assert.assertArrayEquals("EncryptionKey mismatch", encryptionKeyList.get(1),
              resp2.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0).getEncryptionKey().array());
        } else {
          assertNull("MessageMetadata should have been null",
              resp2.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
        }
      } catch (MessageFormatException e) {
        fail();
      }
      releaseNettyBufUnderneathStream(stream);
      // get blob
      ids.clear();
      ids.add(blobIdList.get(0));
      GetRequest getRequest3 =
          new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
      stream = channel3.sendAndReceive(getRequest3).getInputStream();
      GetResponse resp3 = GetResponse.readFrom(stream, clusterMap);
      try {
        BlobData blobData = MessageFormatRecord.deserializeBlob(resp3.getInputStream());
        byte[] blobout = getBlobData(blobData);
        assertArrayEquals(dataList.get(0), blobout);
        if (testEncryption) {
          assertNotNull("MessageMetadata should not have been null",
              resp3.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
          Assert.assertArrayEquals("EncryptionKey mismatch", encryptionKeyList.get(0),
              resp3.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0).getEncryptionKey().array());
        } else {
          assertNull("MessageMetadata should have been null",
              resp3.getPartitionResponseInfoList().get(0).getMessageMetadataList().get(0));
        }
      } catch (MessageFormatException e) {
        fail();
      }
      releaseNettyBufUnderneathStream(stream);
      // get blob all
      ids.clear();
      ids.add(blobIdList.get(0));
      GetRequest getRequest4 =
          new GetRequest(1, "clientid2", MessageFormatFlags.All, partitionRequestInfoList, GetOption.None);
      stream = channel1.sendAndReceive(getRequest4).getInputStream();
      GetResponse resp4 = GetResponse.readFrom(stream, clusterMap);
      try {
        BlobAll blobAll = MessageFormatRecord.deserializeBlobAll(resp4.getInputStream(), blobIdFactory);
        byte[] blobout = getBlobData(blobAll.getBlobData());
        assertArrayEquals(dataList.get(0), blobout);
        if (testEncryption) {
          assertNotNull("MessageMetadata should not have been null", blobAll.getBlobEncryptionKey());
          assertArrayEquals("EncryptionKey mismatch", encryptionKeyList.get(0), blobAll.getBlobEncryptionKey().array());
        } else {
          assertNull("MessageMetadata should have been null", blobAll.getBlobEncryptionKey());
        }
      } catch (MessageFormatException e) {
        fail();
      }
      releaseNettyBufUnderneathStream(stream);
      if (!testEncryption) {
        // get blob data
        // Use router to get the blob
        Properties routerProperties = getRouterProps(routerDatacenter);
        routerProperties.putAll(routerProps);
        VerifiableProperties routerVerifiableProperties = new VerifiableProperties(routerProperties);
        AccountService accountService = new InMemAccountService(false, true);
        Router router = new NonBlockingRouterFactory(routerVerifiableProperties, clusterMap, notificationSystem,
            getSSLFactoryIfRequired(routerVerifiableProperties), accountService).getRouter();
        checkBlobId(router, blobIdList.get(0), dataList.get(0));
        checkBlobId(router, blobIdList.get(1), dataList.get(1));
        checkBlobId(router, blobIdList.get(2), dataList.get(2));
        checkBlobId(router, blobIdList.get(3), dataList.get(3));
        checkBlobId(router, blobIdList.get(4), dataList.get(4));
        checkBlobId(router, blobIdList.get(5), dataList.get(5));
        router.close();
      }

      // fetch blob that does not exist
      // get blob properties
      ids = new ArrayList<BlobId>();
      mockPartitionId =
          (MockPartitionId) clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS).get(0);
      ids.add(new BlobId(CommonTestUtils.getCurrentBlobIdVersion(), BlobId.BlobIdType.NATIVE,
          clusterMap.getLocalDatacenterId(), propertyList.get(0).getAccountId(), propertyList.get(0).getContainerId(),
          mockPartitionId, false, BlobId.BlobDataType.DATACHUNK));
      partitionRequestInfoList.clear();
      partitionRequestInfo = new PartitionRequestInfo(mockPartitionId, ids);
      partitionRequestInfoList.add(partitionRequestInfo);
      GetRequest getRequest5 =
          new GetRequest(1, "clientid2", MessageFormatFlags.BlobProperties, partitionRequestInfoList, GetOption.None);
      stream = channel3.sendAndReceive(getRequest5).getInputStream();
      GetResponse resp5 = GetResponse.readFrom(stream, clusterMap);
      Assert.assertEquals(ServerErrorCode.NoError, resp5.getError());
      Assert.assertEquals(ServerErrorCode.BlobNotFound, resp5.getPartitionResponseInfoList().get(0).getErrorCode());
      releaseNettyBufUnderneathStream(stream);
      // delete a blob and ensure it is propagated
      DeleteRequest deleteRequest = new DeleteRequest(1, "reptest", blobIdList.get(0), System.currentTimeMillis());
      expectedTokenSize += getUpdateRecordSize(blobIdList.get(0), SubRecord.Type.DELETE);
      DataInputStream deleteResponseStream = channel1.sendAndReceive(deleteRequest).getInputStream();
      DeleteResponse deleteResponse = DeleteResponse.readFrom(deleteResponseStream);
      releaseNettyBufUnderneathStream(deleteResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, deleteResponse.getError());

      notificationSystem.awaitBlobDeletions(blobIdList.get(0).getID());
      ids = new ArrayList<BlobId>();
      ids.add(blobIdList.get(0));
      partitionRequestInfoList.clear();
      partitionRequestInfo = new PartitionRequestInfo(partition, ids);
      partitionRequestInfoList.add(partitionRequestInfo);
      GetRequest getRequest6 =
          new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
      stream = channel3.sendAndReceive(getRequest6).getInputStream();
      GetResponse resp6 = GetResponse.readFrom(stream, clusterMap);
      Assert.assertEquals(ServerErrorCode.NoError, resp6.getError());
      Assert.assertEquals(ServerErrorCode.BlobDeleted, resp6.getPartitionResponseInfoList().get(0).getErrorCode());
      releaseNettyBufUnderneathStream(stream);
      // get the data node to inspect replication tokens on
      DataNodeId dataNodeId = clusterMap.getDataNodeId("localhost", interestedDataNodePortNumber);
      checkReplicaTokens(clusterMap, dataNodeId,
          expectedTokenSize - getUpdateRecordSize(blobIdList.get(0), SubRecord.Type.DELETE), "0");

      // Shut down server 1
      cluster.getServers().get(0).shutdown();
      cluster.getServers().get(0).awaitShutdown();
      // Add more data to server 2 and server 3. Recover server 1 and ensure it is completely replicated
      // put blob 7
      putRequest2 = new PutRequest(1, "client1", blobIdList.get(6), propertyList.get(6), ByteBuffer.wrap(usermetadata),
          Unpooled.wrappedBuffer(dataList.get(6)), propertyList.get(6).getBlobSize(), BlobType.DataBlob,
          encryptionKeyList.get(6) != null ? ByteBuffer.wrap(encryptionKeyList.get(6)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(6), blobIdList.get(6),
          encryptionKeyList.get(6) != null ? ByteBuffer.wrap(encryptionKeyList.get(6)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(6));
      putResponseStream = channel2.sendAndReceive(putRequest2).getInputStream();
      response2 = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response2.getError());

      // put blob 8
      putRequest3 = new PutRequest(1, "client1", blobIdList.get(7), propertyList.get(7), ByteBuffer.wrap(usermetadata),
          Unpooled.wrappedBuffer(dataList.get(7)), propertyList.get(7).getBlobSize(), BlobType.DataBlob,
          encryptionKeyList.get(7) != null ? ByteBuffer.wrap(encryptionKeyList.get(7)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(7), blobIdList.get(7),
          encryptionKeyList.get(7) != null ? ByteBuffer.wrap(encryptionKeyList.get(7)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(7));
      putResponseStream = channel3.sendAndReceive(putRequest3).getInputStream();
      response3 = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response3.getError());

      // put blob 9
      putRequest2 = new PutRequest(1, "client1", blobIdList.get(8), propertyList.get(8), ByteBuffer.wrap(usermetadata),
          Unpooled.wrappedBuffer(dataList.get(8)), propertyList.get(8).getBlobSize(), BlobType.DataBlob,
          encryptionKeyList.get(8) != null ? ByteBuffer.wrap(encryptionKeyList.get(8)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(8), blobIdList.get(8),
          encryptionKeyList.get(8) != null ? ByteBuffer.wrap(encryptionKeyList.get(8)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(8));
      putResponseStream = channel2.sendAndReceive(putRequest2).getInputStream();
      response2 = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response2.getError());

      // put blob 10
      putRequest3 = new PutRequest(1, "client1", blobIdList.get(9), propertyList.get(9), ByteBuffer.wrap(usermetadata),
          Unpooled.wrappedBuffer(dataList.get(9)), propertyList.get(9).getBlobSize(), BlobType.DataBlob,
          encryptionKeyList.get(9) != null ? ByteBuffer.wrap(encryptionKeyList.get(9)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(9), blobIdList.get(9),
          encryptionKeyList.get(9) != null ? ByteBuffer.wrap(encryptionKeyList.get(9)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(9));
      putResponseStream = channel3.sendAndReceive(putRequest3).getInputStream();
      response3 = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response3.getError());

      // put blob 11
      putRequest2 =
          new PutRequest(1, "client1", blobIdList.get(10), propertyList.get(10), ByteBuffer.wrap(usermetadata),
              Unpooled.wrappedBuffer(dataList.get(10)), propertyList.get(10).getBlobSize(), BlobType.DataBlob,
              encryptionKeyList.get(10) != null ? ByteBuffer.wrap(encryptionKeyList.get(10)) : null);
      expectedTokenSize += getPutRecordSize(propertyList.get(10), blobIdList.get(10),
          encryptionKeyList.get(10) != null ? ByteBuffer.wrap(encryptionKeyList.get(10)) : null,
          ByteBuffer.wrap(usermetadata), dataList.get(10));
      putResponseStream = channel2.sendAndReceive(putRequest2).getInputStream();
      response2 = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response2.getError());

      checkTtlUpdateStatus(channel2, clusterMap, blobIdFactory, blobIdList.get(10), dataList.get(10), false,
          getExpiryTimeMs(propertyList.get(10)));
      updateBlobTtl(channel2, blobIdList.get(10), cluster.time.milliseconds());
      expectedTokenSize += getUpdateRecordSize(blobIdList.get(10), SubRecord.Type.TTL_UPDATE);
      checkTtlUpdateStatus(channel2, clusterMap, blobIdFactory, blobIdList.get(10), dataList.get(10), true,
          Utils.Infinite_Time);

      cluster.reinitServer(0);
      // wait for server to recover
      notificationSystem.awaitBlobCreations(blobIdList.get(6).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(7).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(8).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(9).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(10).getID());
      notificationSystem.awaitBlobUpdates(blobIdList.get(10).getID(), UpdateType.TTL_UPDATE);
      channel1.disconnect();
      channel1.connect();

      // check all ids exist on server 1
      // get blob
      try {
        checkBlobContent(clusterMap, blobIdList.get(1), channel1, dataList.get(1), encryptionKeyList.get(1));
        checkBlobContent(clusterMap, blobIdList.get(2), channel1, dataList.get(2), encryptionKeyList.get(2));
        checkBlobContent(clusterMap, blobIdList.get(3), channel1, dataList.get(3), encryptionKeyList.get(3));
        checkBlobContent(clusterMap, blobIdList.get(4), channel1, dataList.get(4), encryptionKeyList.get(4));
        checkBlobContent(clusterMap, blobIdList.get(5), channel1, dataList.get(5), encryptionKeyList.get(5));
        checkBlobContent(clusterMap, blobIdList.get(6), channel1, dataList.get(6), encryptionKeyList.get(6));
        checkBlobContent(clusterMap, blobIdList.get(7), channel1, dataList.get(7), encryptionKeyList.get(7));
        checkBlobContent(clusterMap, blobIdList.get(8), channel1, dataList.get(8), encryptionKeyList.get(8));
        checkBlobContent(clusterMap, blobIdList.get(9), channel1, dataList.get(9), encryptionKeyList.get(9));
        checkBlobContent(clusterMap, blobIdList.get(10), channel1, dataList.get(10), encryptionKeyList.get(10));
      } catch (MessageFormatException e) {
        fail();
      }

      // check that the ttl update went through
      checkTtlUpdateStatus(channel1, clusterMap, blobIdFactory, blobIdList.get(10), dataList.get(10), true,
          Utils.Infinite_Time);

      // Shutdown server 1. Remove all its data from all mount path. Recover server 1 and ensure node is built
      cluster.getServers().get(0).shutdown();
      cluster.getServers().get(0).awaitShutdown();

      File mountFile = new File(clusterMap.getReplicaIds(dataNodeId).get(0).getMountPath());
      for (File toDelete : Objects.requireNonNull(mountFile.listFiles())) {
        deleteFolderContent(toDelete, true);
      }
      notificationSystem.decrementCreatedReplica(blobIdList.get(1).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementCreatedReplica(blobIdList.get(2).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementCreatedReplica(blobIdList.get(3).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementCreatedReplica(blobIdList.get(4).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementCreatedReplica(blobIdList.get(5).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementUpdatedReplica(blobIdList.get(5).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort(), UpdateType.TTL_UPDATE);
      notificationSystem.decrementCreatedReplica(blobIdList.get(6).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementCreatedReplica(blobIdList.get(7).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementCreatedReplica(blobIdList.get(8).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementCreatedReplica(blobIdList.get(9).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementCreatedReplica(blobIdList.get(10).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort());
      notificationSystem.decrementUpdatedReplica(blobIdList.get(10).getID(), dataNodeId.getHostname(),
          dataNodeId.getPort(), UpdateType.TTL_UPDATE);

      cluster.reinitServer(0);
      notificationSystem.awaitBlobCreations(blobIdList.get(1).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(2).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(3).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(4).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(5).getID());
      notificationSystem.awaitBlobUpdates(blobIdList.get(5).getID(), UpdateType.TTL_UPDATE);
      notificationSystem.awaitBlobCreations(blobIdList.get(6).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(7).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(8).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(9).getID());
      notificationSystem.awaitBlobCreations(blobIdList.get(10).getID());
      notificationSystem.awaitBlobUpdates(blobIdList.get(10).getID(), UpdateType.TTL_UPDATE);

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

      // check all ids exist on server 1
      // get blob
      try {
        checkBlobContent(clusterMap, blobIdList.get(1), channel1, dataList.get(1), encryptionKeyList.get(1));
        checkBlobContent(clusterMap, blobIdList.get(2), channel1, dataList.get(2), encryptionKeyList.get(2));
        checkBlobContent(clusterMap, blobIdList.get(3), channel1, dataList.get(3), encryptionKeyList.get(3));
        checkBlobContent(clusterMap, blobIdList.get(4), channel1, dataList.get(4), encryptionKeyList.get(4));
        checkBlobContent(clusterMap, blobIdList.get(5), channel1, dataList.get(5), encryptionKeyList.get(5));
        checkBlobContent(clusterMap, blobIdList.get(6), channel1, dataList.get(6), encryptionKeyList.get(6));
        checkBlobContent(clusterMap, blobIdList.get(7), channel1, dataList.get(7), encryptionKeyList.get(7));
        checkBlobContent(clusterMap, blobIdList.get(8), channel1, dataList.get(8), encryptionKeyList.get(8));
        checkBlobContent(clusterMap, blobIdList.get(9), channel1, dataList.get(9), encryptionKeyList.get(9));
        checkBlobContent(clusterMap, blobIdList.get(10), channel1, dataList.get(10), encryptionKeyList.get(10));
      } catch (MessageFormatException e) {
        fail();
      }

      // check that the ttl updates are present
      checkTtlUpdateStatus(channel1, clusterMap, blobIdFactory, blobIdList.get(5), dataList.get(5), true,
          Utils.Infinite_Time);
      checkTtlUpdateStatus(channel1, clusterMap, blobIdFactory, blobIdList.get(10), dataList.get(10), true,
          Utils.Infinite_Time);

      channel1.disconnect();
      channel2.disconnect();
      channel3.disconnect();
    } catch (Exception e) {
      e.printStackTrace();
      fail();
    }
  }
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