@Override
    public void run() {
      while (!putComplete.get() || getCompletedCount.get() < totalPutCount.get()) {
        try {
          int size = generatedBlobAccessInfos.size();
          if (size > 0) {
            long startTimeInMs = SystemTime.getInstance().milliseconds();
            List<Pair<BlobAccessInfo, Integer>> blobInfoBurstCountPairList = new ArrayList<>();
            List<String> toBeDeletedBlobsPerBatch = new ArrayList<>();
            int blobCountPerBatch = Math.min(size, parallelGetCount);
            for (int i = 0; i < blobCountPerBatch; i++) {
              size = generatedBlobAccessInfos.size();
              BlobAccessInfo blobAccessInfo = generatedBlobAccessInfos.get(threadLocalRandom.nextInt(size));
              if (blobAccessInfo != null) {
                int burstCount = threadLocalRandom.nextInt(maxBurstCountPerblob) + 1;
                blobInfoBurstCountPairList.add(new Pair(blobAccessInfo, burstCount));
                blobAccessInfo.referenceCount.addAndGet(burstCount);
                if (blobAccessInfo.referenceCount.get() >= maxGetCountPerBlob) {
                  generatedBlobAccessInfos.remove(blobAccessInfo);
                  getCompletedCount.incrementAndGet();
                  toBeDeletedBlobsPerBatch.add(blobAccessInfo.blobId);
                }
              }
            }

            logger.info("Get thread fetching {} blobs for this phase ", blobCountPerBatch);
            final CountDownLatch countDownLatchForBatch = new CountDownLatch(blobCountPerBatch);
            final AtomicInteger failureCount = new AtomicInteger(0);
            for (Pair<BlobAccessInfo, Integer> blobAccessInfoBurstCountPair : blobInfoBurstCountPairList) {
              int burstCount = blobAccessInfoBurstCountPair.getSecond();
              final BlobAccessInfo blobAccessInfo = blobAccessInfoBurstCountPair.getFirst();
              logger.trace("Get Thread fetching {} {} times ", blobAccessInfo.blobId, burstCount);
              final CountDownLatch countDownLatch = new CountDownLatch(burstCount);
              for (int j = 0; j < burstCount; j++) {
                Callback<Void> callback = new Callback<Void>() {

                  @Override
                  public void onCompletion(Void result, Exception exception) {
                    countDownLatch.countDown();
                    blobAccessInfo.burstGetReferenceCount.decrementAndGet();
                    if (countDownLatch.getCount() == 0) {
                      countDownLatchForBatch.countDown();
                    }
                    if (exception != null) {
                      logger.error("Get and validation of {} failed with ", blobAccessInfo.blobId, exception);
                      failureCount.incrementAndGet();
                    }
                  }
                };
                blobAccessInfo.burstGetReferenceCount.incrementAndGet();
                getHelper.getBlobAndValidate(blobAccessInfo.blobId, blobAccessInfo.blobContent,
                    getHelper.getErrorCodeForNoError(), callback, metricsCollector);
              }
            }
            countDownLatchForBatch.await(5, TimeUnit.MINUTES);
            metricsCollector.updateTimePassedSoFar(SystemTime.getInstance().milliseconds() - startTimeInMs);
            if (deleteOnExit) {
              deleteBlobIds.addAll(toBeDeletedBlobsPerBatch);
            }
            if (failureCount.get() > maxFailuresPerGetBatch) {
              logger.error("{}  failures during this batch, hence exiting the GetThread", failureCount.get());
              break;
            }
          }
          Thread.sleep(threadLocalRandom.nextLong(sleepTimeBetweenBatchGetsInMs));
        } catch (InterruptedException ex) {
          logger.error("InterruptedException in GetThread ", ex);
        }
      }
      metricsCollector.reportMetrics();
      logger.info("Exiting GetThread ");
    }
Get threads to assist in consuming blobs from ambry and verifying its content
Get thread also proceeds in multiple phases until {@code totalPutCount} number of blobs have been fetched and
verified. Blobs and its metadata information are fetched from {@code generatedBlobAccessInfos}. On each phase,
the get thread fetches {@code parallelGetCount} number of blobs from the {@code generatedBlobAccessInfos} and
issues random (max of {@code maxBurstCountPerblob} number of burst GET calls(and validates) to Ambry via
{@link PutGetHelper#getBlobAndValidate(String, byte[], Object, Callback, MetricsCollector)}. Each blob is fetched for a maximum of
{@code maxGetCountPerBlob} times. On reaching the threshold, they are removed from the
{@code generatedBlobAccessInfos} and deletion is called via
{@link PutGetHelper#deleteBlobAndValidate(String, Object, Callback, MetricsCollector)} if {@code deleteOnExit} is set. On completion
of one phase, the thread proceeds onto the next phase after sleeping for max of {@code sleepTimeBetweenBatchGetsInMs} ms.
ambry-tools/src/main/java/com/github/ambry/tools/admin/ConcurrencyTestTool.java