@Override
  public void run() {
    try {
      List<PartitionRequestInfo> partitionRequestInfoList = new ArrayList<PartitionRequestInfo>();
      while (requestsVerified.get() != totalRequests.get() && !cancelTest.get()) {
        Payload payload = payloadQueue.poll(1000, TimeUnit.MILLISECONDS);
        if (payload != null) {
          notificationSystem.awaitBlobCreations(payload.blobId);
          for (MockDataNodeId dataNodeId : clusterMap.getDataNodes()) {
            ConnectedChannel channel1 = null;
            try {
              BlobId blobId = new BlobId(payload.blobId, clusterMap);
              Port port =
                  new Port(portType == PortType.PLAINTEXT ? dataNodeId.getPort() : dataNodeId.getSSLPort(), portType);
              channel1 = connectionPool.checkOutConnection("localhost", port, 10000);
              ArrayList<BlobId> ids = new ArrayList<BlobId>();
              ids.add(blobId);
              partitionRequestInfoList.clear();
              PartitionRequestInfo partitionRequestInfo = new PartitionRequestInfo(ids.get(0).getPartition(), ids);
              partitionRequestInfoList.add(partitionRequestInfo);
              GetRequest getRequest =
                  new GetRequest(1, "clientid2", MessageFormatFlags.BlobProperties, partitionRequestInfoList,
                      GetOption.None);
              channel1.send(getRequest);
              DataInputStream stream = channel1.receive().getInputStream();
              GetResponse resp = GetResponse.readFrom(stream, clusterMap);
              if (resp.getError() != ServerErrorCode.NoError) {
                System.out.println(dataNodeId.getHostname() + " " + dataNodeId.getPort() + " " + resp.getError());
                throw new IllegalStateException();
              } else {
                try {
                  BlobProperties propertyOutput = MessageFormatRecord.deserializeBlobProperties(resp.getInputStream());
                  if (propertyOutput.getBlobSize() != payload.blobProperties.getBlobSize()) {
                    String exceptionMsg =
                        "blob size not matching " + " expected " + payload.blobProperties.getBlobSize() + " actual "
                            + propertyOutput.getBlobSize();
                    System.out.println(exceptionMsg);
                    throw new IllegalStateException(exceptionMsg);
                  }
                  if (!propertyOutput.getServiceId().equals(payload.blobProperties.getServiceId())) {
                    String exceptionMsg =
                        "service id not matching " + " expected " + payload.blobProperties.getServiceId() + " actual "
                            + propertyOutput.getBlobSize();
                    System.out.println(exceptionMsg);
                    throw new IllegalStateException(exceptionMsg);
                  }
                  if (propertyOutput.getAccountId() != payload.blobProperties.getAccountId()) {
                    String exceptionMsg =
                        "accountid not matching " + " expected " + payload.blobProperties.getAccountId() + " actual "
                            + propertyOutput.getAccountId();
                    System.out.println(exceptionMsg);
                    throw new IllegalStateException(exceptionMsg);
                  }
                  if (propertyOutput.getContainerId() != payload.blobProperties.getContainerId()) {
                    String exceptionMsg =
                        "containerId not matching " + " expected " + payload.blobProperties.getContainerId()
                            + " actual " + propertyOutput.getContainerId();
                    System.out.println(exceptionMsg);
                    throw new IllegalStateException(exceptionMsg);
                  }
                  if (propertyOutput.isEncrypted() != payload.blobProperties.isEncrypted()) {
                    String exceptionMsg =
                        "IsEncrypted not matching " + " expected " + payload.blobProperties.isEncrypted() + " actual "
                            + propertyOutput.isEncrypted();
                    System.out.println(exceptionMsg);
                    throw new IllegalStateException(exceptionMsg);
                  }
                  long actualExpiryTimeMs =
                      resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs();
                  checkExpiryTimeMatch(payload, actualExpiryTimeMs, "messageinfo in blobproperty");
                } catch (MessageFormatException e) {
                  e.printStackTrace();
                  throw new IllegalStateException(e);
                }
              }

              // get user metadata
              ids.clear();
              ids.add(blobId);
              partitionRequestInfoList.clear();
              partitionRequestInfo = new PartitionRequestInfo(ids.get(0).getPartition(), ids);
              partitionRequestInfoList.add(partitionRequestInfo);
              getRequest = new GetRequest(1, "clientid2", MessageFormatFlags.BlobUserMetadata, partitionRequestInfoList,
                  GetOption.None);
              channel1.send(getRequest);
              stream = channel1.receive().getInputStream();
              resp = GetResponse.readFrom(stream, clusterMap);
              if (resp.getError() != ServerErrorCode.NoError) {
                System.out.println("Error after get user metadata " + resp.getError());
                throw new IllegalStateException();
              } else {
                try {
                  ByteBuffer userMetadataOutput = MessageFormatRecord.deserializeUserMetadata(resp.getInputStream());
                  if (userMetadataOutput.compareTo(ByteBuffer.wrap(payload.metadata)) != 0) {
                    throw new IllegalStateException();
                  }
                  long actualExpiryTimeMs =
                      resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs();
                  checkExpiryTimeMatch(payload, actualExpiryTimeMs, "messageinfo in usermetadatga");
                } catch (MessageFormatException e) {
                  e.printStackTrace();
                  throw new IllegalStateException();
                }
              }

              // get blob
              ids.clear();
              ids.add(blobId);
              partitionRequestInfoList.clear();
              partitionRequestInfo = new PartitionRequestInfo(ids.get(0).getPartition(), ids);
              partitionRequestInfoList.add(partitionRequestInfo);
              getRequest =
                  new GetRequest(1, "clientid2", MessageFormatFlags.Blob, partitionRequestInfoList, GetOption.None);
              channel1.send(getRequest);
              stream = channel1.receive().getInputStream();
              resp = GetResponse.readFrom(stream, clusterMap);
              //System.out.println("response from get " + resp.getError());
              if (resp.getError() != ServerErrorCode.NoError) {
                System.out.println("Error after get blob " + resp.getError());
                throw new IllegalStateException();
              } else {
                try {
                  BlobData blobData = MessageFormatRecord.deserializeBlob(resp.getInputStream());
                  byte[] blobout = new byte[(int) blobData.getSize()];
                  ByteBuf buffer = blobData.content();
                  try {
                    buffer.readBytes(blobout);
                  } finally {
                    buffer.release();
                  }
                  if (ByteBuffer.wrap(blobout).compareTo(ByteBuffer.wrap(payload.blob)) != 0) {
                    throw new IllegalStateException();
                  }
                  long actualExpiryTimeMs =
                      resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs();
                  checkExpiryTimeMatch(payload, actualExpiryTimeMs, "messageinfo in blobdata");
                } catch (MessageFormatException e) {
                  e.printStackTrace();
                  throw new IllegalStateException();
                }
              }

              // get blob all
              getRequest =
                  new GetRequest(1, "clientid2", MessageFormatFlags.All, partitionRequestInfoList, GetOption.None);
              channel1.send(getRequest);
              stream = channel1.receive().getInputStream();
              resp = GetResponse.readFrom(stream, clusterMap);
              if (resp.getError() != ServerErrorCode.NoError) {
                System.out.println("Error after get blob " + resp.getError());
                throw new IllegalStateException();
              } else {
                try {
                  BlobAll blobAll =
                      MessageFormatRecord.deserializeBlobAll(resp.getInputStream(), new BlobIdFactory(clusterMap));
                  byte[] blobout = new byte[(int) blobAll.getBlobData().getSize()];
                  ByteBuf buffer = blobAll.getBlobData().content();
                  try {
                    buffer.readBytes(blobout);
                  } finally {
                    buffer.release();
                  }
                  if (ByteBuffer.wrap(blobout).compareTo(ByteBuffer.wrap(payload.blob)) != 0) {
                    throw new IllegalStateException();
                  }
                  long actualExpiryTimeMs =
                      resp.getPartitionResponseInfoList().get(0).getMessageInfoList().get(0).getExpirationTimeInMs();
                  checkExpiryTimeMatch(payload, actualExpiryTimeMs, "messageinfo in bloball");
                } catch (MessageFormatException e) {
                  e.printStackTrace();
                  throw new IllegalStateException();
                }
              }

              if (payload.blobProperties.getTimeToLiveInSeconds() != Utils.Infinite_Time) {
                // ttl update, check and wait for replication
                ServerTestUtil.updateBlobTtl(channel1, new BlobId(payload.blobId, clusterMap), time.milliseconds());
                ServerTestUtil.checkTtlUpdateStatus(channel1, clusterMap, new BlobIdFactory(clusterMap), blobId,
                    payload.blob, true, Utils.Infinite_Time);
                notificationSystem.awaitBlobUpdates(payload.blobId, UpdateType.TTL_UPDATE);
                BlobProperties old = payload.blobProperties;
                payload.blobProperties =
                    new BlobProperties(old.getBlobSize(), old.getServiceId(), old.getOwnerId(), old.getContentType(),
                        old.isEncrypted(), Utils.Infinite_Time, old.getCreationTimeInMs(), old.getAccountId(),
                        old.getContainerId(), old.isEncrypted(), old.getExternalAssetTag(), old.getContentEncoding(),
                        old.getFilename(), old.getReservedMetadataBlobId());
              }
            } catch (Exception e) {
              if (channel1 != null) {
                connectionPool.destroyConnection(channel1);
                channel1 = null;
              }
            } finally {
              if (channel1 != null) {
                connectionPool.checkInConnection(channel1);
                channel1 = null;
              }
            }
          }
          requestsVerified.incrementAndGet();
        }
      }
    } catch (Exception e) {
      e.printStackTrace();
      cancelTest.set(true);
    } finally {
      completedLatch.countDown();
    }
  }
Verifier thread to get blob from ambry and check with the original blob
ambry-test-utils/src/main/java/com/github/ambry/server/Verifier.java