static void undeleteCornerCasesTest(MockCluster cluster, PortType portType, SSLConfig clientSSLConfig1,
      SSLConfig clientSSLConfig2, SSLConfig clientSSLConfig3, SSLSocketFactory clientSSLSocketFactory1,
      SSLSocketFactory clientSSLSocketFactory2, SSLSocketFactory clientSSLSocketFactory3,
      MockNotificationSystem notificationSystem, Properties routerProps, boolean testEncryption) {
    MockClusterMap clusterMap = cluster.getClusterMap();
    byte[] userMetadata = new byte[1000];
    byte[] data = new byte[31870];
    byte[] encryptionKey = new byte[100];
    short accountId = Utils.getRandomShort(TestUtils.RANDOM);
    short containerId = Utils.getRandomShort(TestUtils.RANDOM);
    BlobProperties properties =
        new BlobProperties(31870, "serviceid1", accountId, containerId, testEncryption, cluster.time.milliseconds());
    TestUtils.RANDOM.nextBytes(userMetadata);
    TestUtils.RANDOM.nextBytes(data);
    if (testEncryption) {
      TestUtils.RANDOM.nextBytes(encryptionKey);
    }
    short blobIdVersion = CommonTestUtils.getCurrentBlobIdVersion();
    Map<String, List<DataNodeId>> dataNodesPerDC =
        clusterMap.getDataNodes().stream().collect(Collectors.groupingBy(DataNodeId::getDatacenterName));
    Map<String, Pair<SSLConfig, SSLSocketFactory>> sslSettingPerDC = new HashMap<>();
    sslSettingPerDC.put("DC1", new Pair<>(clientSSLConfig1, clientSSLSocketFactory1));
    sslSettingPerDC.put("DC2", new Pair<>(clientSSLConfig2, clientSSLSocketFactory2));
    sslSettingPerDC.put("DC3", new Pair<>(clientSSLConfig3, clientSSLSocketFactory3));

    List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS);
    DataNodeId dataNodeId = dataNodesPerDC.get("DC1").get(0);
    Router router = null;
    try {
      Properties routerProperties = getRouterProps("DC1");
      routerProperties.putAll(routerProps);
      VerifiableProperties routerVerifiableProps = new VerifiableProperties(routerProperties);
      AccountService accountService = new InMemAccountService(false, true);
      router = new NonBlockingRouterFactory(routerVerifiableProps, clusterMap, new MockNotificationSystem(clusterMap),
          getSSLFactoryIfRequired(routerVerifiableProps), accountService).getRouter();

      // channels to all datanodes
      List<ConnectedChannel> channels = new ArrayList<>();
      for (Map.Entry<String, List<DataNodeId>> entry : dataNodesPerDC.entrySet()) {
        Pair<SSLConfig, SSLSocketFactory> pair = sslSettingPerDC.get(entry.getKey());
        for (DataNodeId node : entry.getValue()) {
          ConnectedChannel connectedChannel =
              getBlockingChannelBasedOnPortType(portType, node, pair.getSecond(), pair.getFirst());
          connectedChannel.connect();
          channels.add(connectedChannel);
        }
      }

      //////////////////////////////////////////////////////
      // Corner case 1: When only one datacenter has delete
      //////////////////////////////////////////////////////
      BlobId blobId1 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionIds.get(0), false,
          BlobId.BlobDataType.DATACHUNK);
      ConnectedChannel channel =
          getBlockingChannelBasedOnPortType(portType, dataNodeId, clientSSLSocketFactory1, clientSSLConfig1);
      channel.connect();

      PutRequest putRequest =
          new PutRequest(1, "client1", blobId1, properties, ByteBuffer.wrap(userMetadata), Unpooled.wrappedBuffer(data),
              properties.getBlobSize(), BlobType.DataBlob, testEncryption ? ByteBuffer.wrap(encryptionKey) : null);

      DataInputStream putResponseStream = channel.sendAndReceive(putRequest).getInputStream();

      PutResponse response = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response.getError());

      notificationSystem.awaitBlobCreations(blobId1.toString());

      // Now stop the replications this partition.
      PartitionId partitionId = blobId1.getPartition();
      controlReplicationForPartition(channels, partitionId, false);

      // Now send the delete to two data nodes in the same DC
      List<DataNodeId> toBeDeleteDataNodes = dataNodesPerDC.values().stream().findFirst().get();
      Pair<SSLConfig, SSLSocketFactory> pair = sslSettingPerDC.get(toBeDeleteDataNodes.get(0).getDatacenterName());
      ConnectedChannel channel1 =
          getBlockingChannelBasedOnPortType(portType, toBeDeleteDataNodes.get(0), pair.getSecond(), pair.getFirst());
      channel1.connect();
      ConnectedChannel channel2 =
          getBlockingChannelBasedOnPortType(portType, toBeDeleteDataNodes.get(1), pair.getSecond(), pair.getFirst());
      channel2.connect();
      DeleteRequest deleteRequest1 = new DeleteRequest(1, "deleteClient", blobId1, System.currentTimeMillis());
      DataInputStream stream = channel1.sendAndReceive(deleteRequest1).getInputStream();
      DeleteResponse deleteResponse = DeleteResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, deleteResponse.getError());
      DeleteRequest deleteRequest2 =
          new DeleteRequest(1, "deleteClient", blobId1, deleteRequest1.getDeletionTimeInMs());
      stream = channel2.sendAndReceive(deleteRequest2).getInputStream();
      deleteResponse = DeleteResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, deleteResponse.getError());

      // Now send the undelete operation through router, and it should fail because of not deleted error.
      Future<Void> future = router.undeleteBlob(blobId1.toString(), "service");
      try {
        future.get();
        fail("Undelete blob " + blobId1.toString() + " should fail");
      } catch (ExecutionException e) {
        assertTrue(e.getCause() instanceof RouterException);
        assertEquals(RouterErrorCode.BlobNotDeleted, ((RouterException) e.getCause()).getErrorCode());
      }

      // Now see if either data node 1 or data node 2 has undelete or not, if so, undelete would replicate. If not,
      // delete would replicate.
      List<PartitionRequestInfo> partitionRequestInfoList = getPartitionRequestInfoListFromBlobId(blobId1);
      boolean hasUndelete = false;
      for (ConnectedChannel connectedChannel : new ConnectedChannel[]{channel1, channel2}) {
        GetRequest getRequest =
            new GetRequest(1, "clientId1", MessageFormatFlags.BlobProperties, partitionRequestInfoList,
                GetOption.Include_All);
        stream = channel1.sendAndReceive(getRequest).getInputStream();
        GetResponse getResponse = GetResponse.readFrom(stream, clusterMap);
        Assert.assertEquals(ServerErrorCode.NoError, getResponse.getPartitionResponseInfoList().get(0).getErrorCode());
        MessageFormatRecord.deserializeBlobProperties(getResponse.getInputStream());
        hasUndelete =
            getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getLifeVersion() == (short) 1;
        if (hasUndelete) {
          break;
        }
      }
      releaseNettyBufUnderneathStream(stream);
      // Now restart the replication
      controlReplicationForPartition(channels, partitionId, true);
      if (hasUndelete) {
        notificationSystem.awaitBlobUndeletes(blobId1.toString());
      } else {
        notificationSystem.awaitBlobDeletions(blobId1.toString());
      }

      for (ConnectedChannel connectedChannel : channels) {
        GetRequest getRequest =
            new GetRequest(1, "clientId1", MessageFormatFlags.BlobProperties, partitionRequestInfoList,
                GetOption.Include_All);
        stream = connectedChannel.sendAndReceive(getRequest).getInputStream();
        GetResponse getResponse = GetResponse.readFrom(stream, clusterMap);
        releaseNettyBufUnderneathStream(stream);
        Assert.assertEquals(ServerErrorCode.NoError, getResponse.getPartitionResponseInfoList().get(0).getErrorCode());
        MessageFormatRecord.deserializeBlobProperties(getResponse.getInputStream());
        if (hasUndelete) {
          Assert.assertEquals((short) 1,
              getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getLifeVersion());
          assertTrue(getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).isUndeleted());
          assertFalse(getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).isDeleted());
        } else {
          Assert.assertEquals((short) 0,
              getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getLifeVersion());
          assertTrue(getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).isDeleted());
          assertFalse(getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).isUndeleted());
        }
      }

      /////////////////////////////////////////////////////////////
      // Corner case 2: two data nodes have different life versions
      ////////////////////////////////////////////////////////////
      BlobId blobId2 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionIds.get(0), false,
          BlobId.BlobDataType.DATACHUNK);
      putRequest =
          new PutRequest(1, "client1", blobId2, properties, ByteBuffer.wrap(userMetadata), Unpooled.wrappedBuffer(data),
              properties.getBlobSize(), BlobType.DataBlob, testEncryption ? ByteBuffer.wrap(encryptionKey) : null);
      putResponseStream = channel.sendAndReceive(putRequest).getInputStream();
      response = PutResponse.readFrom(putResponseStream);
      releaseNettyBufUnderneathStream(putResponseStream);
      Assert.assertEquals(ServerErrorCode.NoError, response.getError());
      notificationSystem.awaitBlobCreations(blobId2.toString());

      // Now delete this blob on all servers.
      DeleteRequest deleteRequest = new DeleteRequest(1, "deleteClient", blobId2, System.currentTimeMillis());
      stream = channel.sendAndReceive(deleteRequest).getInputStream();
      deleteResponse = DeleteResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, deleteResponse.getError());
      notificationSystem.awaitBlobDeletions(blobId2.toString());

      // Now stop the replication
      partitionId = blobId2.getPartition();
      controlReplicationForPartition(channels, partitionId, false);

      // Now send the undelete to two data nodes in the same DC and then send delete
      UndeleteRequest undeleteRequest = new UndeleteRequest(1, "undeleteClient", blobId2, System.currentTimeMillis());
      stream = channel1.sendAndReceive(undeleteRequest).getInputStream();
      UndeleteResponse undeleteResponse = UndeleteResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, undeleteResponse.getError());
      Assert.assertEquals((short) 1, undeleteResponse.getLifeVersion());
      undeleteRequest = new UndeleteRequest(1, "undeleteClient", blobId2, undeleteRequest.getOperationTimeMs());
      stream = channel2.sendAndReceive(undeleteRequest).getInputStream();
      undeleteResponse = UndeleteResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, undeleteResponse.getError());
      Assert.assertEquals((short) 1, undeleteResponse.getLifeVersion());

      deleteRequest1 = new DeleteRequest(1, "deleteClient", blobId2, System.currentTimeMillis());
      stream = channel1.sendAndReceive(deleteRequest1).getInputStream();
      deleteResponse = DeleteResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, deleteResponse.getError());
      deleteRequest2 = new DeleteRequest(1, "deleteClient", blobId2, deleteRequest1.getDeletionTimeInMs());
      stream = channel2.sendAndReceive(deleteRequest2).getInputStream();
      deleteResponse = DeleteResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, deleteResponse.getError());

      // Now send the undelete operation through router, and it should fail because of lifeVersion conflict error.
      future = router.undeleteBlob(blobId2.toString(), "service");
      try {
        future.get();
        fail("Undelete blob " + blobId2.toString() + " should fail");
      } catch (ExecutionException e) {
        assertTrue(e.getCause() instanceof RouterException);
        assertEquals(RouterErrorCode.LifeVersionConflict, ((RouterException) e.getCause()).getErrorCode());
      }

      // Now restart the replication
      controlReplicationForPartition(channels, partitionId, true);
      notificationSystem.awaitBlobUndeletes(blobId2.toString());

      // Now after replication is resumed, the undelete of lifeversion 2 will eventually be replicated to all servers.
      partitionRequestInfoList = getPartitionRequestInfoListFromBlobId(blobId2);
      for (ConnectedChannel connectedChannel : channels) {
        // Even if the notificationSystem acknowledged the undelete, it might be triggered by undelete at lifeversion 1.
        // So check in a loop with a time out.
        long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(10);
        while (true) {
          GetRequest getRequest =
              new GetRequest(1, "clientId1", MessageFormatFlags.BlobProperties, partitionRequestInfoList,
                  GetOption.Include_All);
          stream = connectedChannel.sendAndReceive(getRequest).getInputStream();
          GetResponse getResponse = GetResponse.readFrom(stream, clusterMap);
          Assert.assertEquals(ServerErrorCode.NoError,
              getResponse.getPartitionResponseInfoList().get(0).getErrorCode());
          MessageFormatRecord.deserializeBlobProperties(getResponse.getInputStream());
          if (getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getLifeVersion() == 2) {
            assertTrue(getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).isUndeleted());
            assertFalse(getResponse.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).isDeleted());
            break;
          } else {
            Thread.sleep(1000);
            if (System.currentTimeMillis() > deadline) {
              throw new TimeoutException(
                  "Fail to get blob " + blobId2 + " at lifeversion 2 at " + connectedChannel.getRemoteHost());
            }
          }
        }
      }
      releaseNettyBufUnderneathStream(stream);
      for (ConnectedChannel connectedChannel : channels) {
        connectedChannel.disconnect();
      }
      channel1.disconnect();
      channel2.disconnect();
      channel.disconnect();
    } catch (Exception e) {
      e.printStackTrace();
      fail();
    } finally {
      if (router != null) {
        try {
          router.close();
        } catch (Exception e) {
        }
      }
    }
  }
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