static void endToEndReplicationWithMultiNodeMultiPartitionMultiDCTest(String sourceDatacenter,
      String sslEnabledDatacenters, PortType portType, MockCluster cluster, MockNotificationSystem notificationSystem,
      Properties routerProps) throws Exception {
    Properties props = new Properties();
    props.setProperty("router.hostname", "localhost");
    props.setProperty("router.datacenter.name", sourceDatacenter);
    props.setProperty("router.put.request.parallelism", "1");
    props.setProperty("router.put.success.target", "1");
    props.setProperty("clustermap.cluster.name", "test");
    props.setProperty("clustermap.datacenter.name", sourceDatacenter);
    props.setProperty("clustermap.host.name", "localhost");
    props.setProperty("kms.default.container.key", TestUtils.getRandomKey(32));
    props.putAll(routerProps);
    VerifiableProperties verifiableProperties = new VerifiableProperties(props);
    AccountService accountService = new InMemAccountService(false, true);
    Router router = new NonBlockingRouterFactory(verifiableProperties, cluster.getClusterMap(), notificationSystem,
        getSSLFactoryIfRequired(verifiableProperties), accountService).getRouter();
    int numberOfRequestsToSend = 15;
    int numberOfVerifierThreads = 3;
    final LinkedBlockingQueue<Payload> payloadQueue = new LinkedBlockingQueue<Payload>();
    final AtomicReference<Exception> exceptionRef = new AtomicReference<>(null);
    final CountDownLatch callbackLatch = new CountDownLatch(numberOfRequestsToSend);
    List<Future<String>> putFutures = new ArrayList<>(numberOfRequestsToSend);
    short accountId = Utils.getRandomShort(TestUtils.RANDOM);
    short containerId = Utils.getRandomShort(TestUtils.RANDOM);
    for (int i = 0; i < numberOfRequestsToSend; i++) {
      int size = new Random().nextInt(5000);
      final BlobProperties properties =
          new BlobProperties(size, "service1", "owner id check", "image/jpeg", false, TestUtils.TTL_SECS,
              cluster.time.milliseconds(), accountId, containerId, false, null, null, null, null);
      final byte[] metadata = new byte[new Random().nextInt(1000)];
      final byte[] blob = new byte[size];
      TestUtils.RANDOM.nextBytes(metadata);
      TestUtils.RANDOM.nextBytes(blob);
      Future<String> future =
          router.putBlob(null, properties, metadata, new ByteBufferReadableStreamChannel(ByteBuffer.wrap(blob)),
              new PutBlobOptionsBuilder().build(), new Callback<String>() {
                @Override
                public void onCompletion(String result, Exception exception) {
                  if (exception == null) {
                    payloadQueue.add(new Payload(properties, metadata, blob, result));
                  } else {
                    exceptionRef.set(exception);
                  }
                  callbackLatch.countDown();
                }
              }, QUOTA_CHARGE_EVENT_LISTENER);
      putFutures.add(future);
    }
    for (Future<String> future : putFutures) {
      future.get(20, TimeUnit.SECONDS);
    }
    assertTrue("Did not receive all callbacks in time", callbackLatch.await(1, TimeUnit.SECONDS));
    if (exceptionRef.get() != null) {
      throw exceptionRef.get();
    }
    // put away for future use
    Payload payload1 = payloadQueue.peek();
    MockClusterMap clusterMap = cluster.getClusterMap();
    BlobId blobId1 = new BlobId(payload1.blobId, clusterMap);

    assertEquals("Did not put expected number of blobs", numberOfRequestsToSend, payloadQueue.size());
    Properties sslProps = new Properties();
    sslProps.putAll(routerProps);
    sslProps.setProperty("clustermap.ssl.enabled.datacenters", sslEnabledDatacenters);
    sslProps.setProperty("clustermap.cluster.name", "test");
    sslProps.setProperty("clustermap.datacenter.name", sourceDatacenter);
    sslProps.setProperty("clustermap.host.name", "localhost");
    sslProps.setProperty("connectionpool.read.timeout.ms", "15000");
    VerifiableProperties vProps = new VerifiableProperties(sslProps);
    ConnectionPool connectionPool =
        new BlockingChannelConnectionPool(new ConnectionPoolConfig(vProps), new SSLConfig(vProps),
            new ClusterMapConfig(vProps), new MetricRegistry());
    CountDownLatch verifierLatch = new CountDownLatch(numberOfVerifierThreads);
    AtomicInteger totalRequests = new AtomicInteger(numberOfRequestsToSend);
    AtomicInteger verifiedRequests = new AtomicInteger(0);
    AtomicBoolean cancelTest = new AtomicBoolean(false);
    for (int i = 0; i < numberOfVerifierThreads; i++) {
      Thread thread = new Thread(
          new Verifier(payloadQueue, verifierLatch, totalRequests, verifiedRequests, cluster.getClusterMap(),
              cancelTest, portType, connectionPool, notificationSystem, cluster.time));
      thread.start();
    }
    assertTrue("Did not verify in 2 minutes", verifierLatch.await(2, TimeUnit.MINUTES));
    assertEquals(totalRequests.get(), verifiedRequests.get());

    BlobIdFactory blobIdFactory = new BlobIdFactory(clusterMap);
    MockDataNodeId dataNodeId = clusterMap.getDataNodes().get(0);
    Port port = new Port(portType == PortType.PLAINTEXT ? dataNodeId.getPort() : dataNodeId.getSSLPort(), portType);
    ConnectedChannel channel = connectionPool.checkOutConnection("localhost", port, 10000);
    PartitionId partitionId = blobId1.getPartition();

    // stop the store via AdminRequest
    System.out.println("Begin to stop a BlobStore");
    AdminRequest adminRequest =
        new AdminRequest(AdminRequestOrResponseType.BlobStoreControl, partitionId, 1, "clientid2");
    BlobStoreControlAdminRequest controlRequest =
        new BlobStoreControlAdminRequest((short) 0, BlobStoreControlAction.StopStore, adminRequest);
    DataInputStream stream = channel.sendAndReceive(controlRequest).getInputStream();
    AdminResponse adminResponse = AdminResponse.readFrom(stream);
    releaseNettyBufUnderneathStream(stream);
    Assert.assertEquals("Stop store admin request should succeed", ServerErrorCode.NoError, adminResponse.getError());

    // put a blob on a stopped store, which should fail
    byte[] userMetadata = new byte[1000];
    byte[] data = new byte[3187];
    BlobProperties properties =
        new BlobProperties(3187, "serviceid1", accountId, containerId, false, cluster.time.milliseconds());
    BlobId blobId2 = new BlobId(CommonTestUtils.getCurrentBlobIdVersion(), BlobId.BlobIdType.NATIVE,
        clusterMap.getLocalDatacenterId(), accountId, containerId, partitionId, false, BlobId.BlobDataType.DATACHUNK);
    PutRequest putRequest2 =
        new PutRequest(1, "clientId2", blobId2, properties, ByteBuffer.wrap(userMetadata), Unpooled.wrappedBuffer(data),
            properties.getBlobSize(), BlobType.DataBlob, null);
    DataInputStream putResponseStream = channel.sendAndReceive(putRequest2).getInputStream();
    PutResponse response2 = PutResponse.readFrom(putResponseStream);
    releaseNettyBufUnderneathStream(putResponseStream);
    Assert.assertEquals("Put blob on stopped store should fail", ServerErrorCode.ReplicaUnavailable,
        response2.getError());

    // get a blob properties on a stopped store, which should fail
    ArrayList<BlobId> ids = new ArrayList<>();
    ids.add(blobId1);
    ArrayList<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<>();
    PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(partitionId, ids);
    partitionRequestInfoList.add(partitionRequestInfo);
    GetRequest getRequest1 =
        new GetRequest(1, "clientId1", MessageFormatFlags.BlobProperties, partitionRequestInfoList, GetOption.None);
    stream = channel.sendAndReceive(getRequest1).getInputStream();
    GetResponse resp1 = GetResponse.readFrom(stream, clusterMap);
    Assert.assertEquals("Get blob properties on stopped store should fail", ServerErrorCode.ReplicaUnavailable,
        resp1.getPartitionResponseInfoList().get(0).getErrorCode());
    releaseNettyBufUnderneathStream(stream);
    // delete a blob on a stopped store, which should fail
    DeleteRequest deleteRequest = new DeleteRequest(1, "clientId1", blobId1, System.currentTimeMillis());
    stream = channel.sendAndReceive(deleteRequest).getInputStream();
    DeleteResponse deleteResponse = DeleteResponse.readFrom(stream);
    releaseNettyBufUnderneathStream(stream);
    Assert.assertEquals("Delete blob on stopped store should fail", ServerErrorCode.ReplicaUnavailable,
        deleteResponse.getError());

    // start the store via AdminRequest
    System.out.println("Begin to restart the BlobStore");
    adminRequest = new AdminRequest(AdminRequestOrResponseType.BlobStoreControl, partitionId, 1, "clientId");
    controlRequest = new BlobStoreControlAdminRequest((short) 0, BlobStoreControlAction.StartStore, adminRequest);
    stream = channel.sendAndReceive(controlRequest).getInputStream();
    adminResponse = AdminResponse.readFrom(stream);
    releaseNettyBufUnderneathStream(stream);
    Assert.assertEquals("Start store admin request should succeed", ServerErrorCode.NoError, adminResponse.getError());
    List<? extends ReplicaId> replicaIds = partitionId.getReplicaIds();
    for (ReplicaId replicaId : replicaIds) {
      // forcibly mark replicas and disks as up.
      MockReplicaId mockReplicaId = (MockReplicaId) replicaId;
      mockReplicaId.markReplicaDownStatus(false);
      ((MockDiskId) mockReplicaId.getDiskId()).setDiskState(HardwareState.AVAILABLE, false);
    }

    // put a blob on a restarted store , which should succeed
    putRequest2 =
        new PutRequest(1, "clientId2", blobId2, properties, ByteBuffer.wrap(userMetadata), Unpooled.wrappedBuffer(data),
            properties.getBlobSize(), BlobType.DataBlob, null);
    putResponseStream = channel.sendAndReceive(putRequest2).getInputStream();
    response2 = PutResponse.readFrom(putResponseStream);
    releaseNettyBufUnderneathStream(putResponseStream);
    Assert.assertEquals("Put blob on restarted store should succeed", ServerErrorCode.NoError, response2.getError());
    // verify the put blob has been replicated successfully.
    notificationSystem.awaitBlobCreations(blobId2.getID());

    // get a blob on a restarted store , which should succeed
    ids = new ArrayList<BlobId>();
    ids.add(blobId2);
    partitionRequestInfoList = new ArrayList<PartitionRequestInfo>();
    partitionRequestInfo = new PartitionRequestInfo(partitionId, ids);
    partitionRequestInfoList.add(partitionRequestInfo);
    GetRequest getRequest2 =
        new GetRequest(1, "clientId2", MessageFormatFlags.All, partitionRequestInfoList, GetOption.None);
    stream = channel.sendAndReceive(getRequest2).getInputStream();
    GetResponse resp2 = GetResponse.readFrom(stream, clusterMap);
    InputStream responseStream = resp2.getInputStream();
    BlobAll blobAll = MessageFormatRecord.deserializeBlobAll(responseStream, blobIdFactory);
    byte[] actualBlobData = getBlobData(blobAll.getBlobData());
    assertArrayEquals("Content mismatch.", data, actualBlobData);
    releaseNettyBufUnderneathStream(stream);
    // delete a blob on a restarted store , which should succeed
    deleteRequest = new DeleteRequest(1, "clientId2", blobId2, System.currentTimeMillis());
    stream = channel.sendAndReceive(deleteRequest).getInputStream();
    deleteResponse = DeleteResponse.readFrom(stream);
    releaseNettyBufUnderneathStream(stream);
    Assert.assertEquals("Delete blob on restarted store should succeed", ServerErrorCode.NoError,
        deleteResponse.getError());

    router.close();
    connectionPool.shutdown();
  }
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