static void undeleteRecoveryTest(Port targetPort, MockCluster cluster, SSLConfig clientSSLConfig,
      SSLSocketFactory clientSSLSocketFactory) {
    try {
      MockClusterMap clusterMap = cluster.getClusterMap();
      byte[] userMetadata = new byte[1000];
      byte[] data = new byte[31870];
      short accountId = Utils.getRandomShort(TestUtils.RANDOM);
      short containerId = Utils.getRandomShort(TestUtils.RANDOM);

      BlobProperties properties =
          new BlobProperties(31870, "serviceid1", accountId, containerId, false, cluster.time.milliseconds());
      TestUtils.RANDOM.nextBytes(userMetadata);
      TestUtils.RANDOM.nextBytes(data);
      List<PartitionId> partitionIds = clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS);
      short blobIdVersion = CommonTestUtils.getCurrentBlobIdVersion();

      BlobId blobId1 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          properties.getAccountId(), properties.getContainerId(), partitionIds.get(0), false,
          BlobId.BlobDataType.DATACHUNK);

      // put blob 1
      PutRequest putRequest =
          new PutRequest(1, "client1", blobId1, properties, ByteBuffer.wrap(userMetadata), Unpooled.wrappedBuffer(data),
              properties.getBlobSize(), BlobType.DataBlob, null);
      ConnectedChannel channel =
          getBlockingChannelBasedOnPortType(targetPort, "localhost", clientSSLSocketFactory, clientSSLConfig);
      channel.connect();
      DataInputStream stream = channel.sendAndReceive(putRequest).getInputStream();
      PutResponse response = PutResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, response.getError());

      for (int i = 0; i < 2; i++) {
        // delete blob 1
        deleteBlob(channel, blobId1, cluster.time.milliseconds());
        // undelete blob 1
        undeleteBlob(channel, blobId1, cluster.time.milliseconds(), (short) (i + 1));
      }

      // put blob 2 that is expired (Add additional 5 secs to avoid Blob_Update_Not_Allowed failure as TtlUpdate op time
      // is also cluster.time.milliseconds(). Theoretically, it should succeed as op time = expiry time - buffer time.
      // However, the index value converts ms to sec when putting a blob, so the milliseconds part of initial put blob
      // time is wiped out and makes op time > expiry time - buffer time. Adding some time should avoid this failure.)
      long ttl = 24 * 60 * 60 + 5;
      BlobProperties propertiesExpired =
          new BlobProperties(31870, "serviceid1", "ownerid", "jpeg", false, ttl, cluster.time.milliseconds(), accountId,
              containerId, false, null, null, null, null);
      BlobId blobId2 = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, clusterMap.getLocalDatacenterId(),
          propertiesExpired.getAccountId(), propertiesExpired.getContainerId(), partitionIds.get(0), false,
          BlobId.BlobDataType.DATACHUNK);
      PutRequest putRequest2 = new PutRequest(1, "client1", blobId2, propertiesExpired, ByteBuffer.wrap(userMetadata),
          Unpooled.wrappedBuffer(data), properties.getBlobSize(), BlobType.DataBlob, null);
      stream = channel.sendAndReceive(putRequest2).getInputStream();
      PutResponse response2 = PutResponse.readFrom(stream);
      releaseNettyBufUnderneathStream(stream);
      Assert.assertEquals(ServerErrorCode.NoError, response2.getError());

      for (int i = 0; i < 2; i++) {
        // delete blob 2
        deleteBlob(channel, blobId2, cluster.time.milliseconds());
        // undelete blob 2
        undeleteBlob(channel, blobId2, cluster.time.milliseconds(), (short) (i + 1));
      }

      // ttl update blob 2
      updateBlobTtl(channel, blobId2, cluster.time.milliseconds());
      cluster.time.sleep(ttl + 10000);

      // Now stops the server and remove all the index files for this partition and test its recovery.
      channel.disconnect();
      AmbryServer server = cluster.getServers().get(0);
      server.shutdown();
      server.awaitShutdown();

      MockDataNodeId dataNode = (MockDataNodeId) clusterMap.getDataNodeId("localhost", channel.getRemotePort());
      for (ReplicaId replica : partitionIds.get(0).getReplicaIds()) {
        if (replica.getDataNodeId().equals(dataNode)) {
          for (File file : Objects.requireNonNull(
              new File(replica.getReplicaPath()).listFiles((file, filename) -> filename.endsWith("index")))) {
            file.delete();
          }
        }
      }
      cluster.reinitServer(0);
      channel = getBlockingChannelBasedOnPortType(targetPort, "localhost", clientSSLSocketFactory, clientSSLConfig);
      channel.connect();

      // Now verify that we can fetch blob1 and blob2.
      for (BlobId blobId : new BlobId[]{blobId1, blobId2}) {
        long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(10);

        while (true) {
          // get blob properties
          ArrayList<BlobId> ids = new ArrayList<>();
          ids.add(blobId);
          ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
          PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(partitionIds.get(0), ids);
          partitionRequestInfoList.add(partitionRequestInfo);
          GetRequest getRequest =
              new GetRequest(1, "clientid2", MessageFormatFlags.BlobProperties, partitionRequestInfoList,
                  GetOption.None);
          stream = channel.sendAndReceive(getRequest).getInputStream();
          GetResponse getResponse = GetResponse.readFrom(stream, clusterMap);
          if (getResponse.getPartitionResponseInfoList().get(0).getErrorCode() == ServerErrorCode.NoError) {
            BlobProperties propertyOutput = MessageFormatRecord.deserializeBlobProperties(getResponse.getInputStream());
            assertEquals(31870, propertyOutput.getBlobSize());
            assertEquals("serviceid1", propertyOutput.getServiceId());
            assertEquals("AccountId mismatch", accountId, propertyOutput.getAccountId());
            assertEquals("ContainerId mismatch", containerId, propertyOutput.getContainerId());
            releaseNettyBufUnderneathStream(stream);
            break;
          } else {
            Thread.sleep(1000);
            if (System.currentTimeMillis() > deadline) {
              throw new TimeoutException("Fail to get blob " + blobId + " at  " + channel.getRemoteHost());
            }
          }
        }
      }
      channel.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