@Override
    public void run() {
      try {
        while (currentPutCount.get() < totalPutCount) {
          long startTimeInMs = SystemTime.getInstance().milliseconds();
          int burstCount = threadLocalRandom.nextInt(maxParallelRequest) + 1;
          logger.info("PutThread producing {} times ", burstCount);
          final CountDownLatch countDownLatch = new CountDownLatch(burstCount);
          final AtomicInteger failureCount = new AtomicInteger(0);
          for (int j = 0; j < burstCount; j++) {
            Callback<Pair<String, byte[]>> callback = new Callback<Pair<String, byte[]>>() {

              @Override
              public void onCompletion(Pair<String, byte[]> result, Exception exception) {
                if (exception == null) {
                  if (currentPutCount.get() < totalPutCount) {
                    generatedBlobAccessInfos.add(new BlobAccessInfo(result.getFirst(), result.getSecond()));
                    currentPutCount.incrementAndGet();
                  }
                } else {
                  logger.error("PutBlob failed with ", exception);
                  failureCount.incrementAndGet();
                }
                countDownLatch.countDown();
              }
            };
            putHelper.putBlob(callback, metricsCollector);
          }
          countDownLatch.await(5, TimeUnit.MINUTES);
          metricsCollector.updateTimePassedSoFar(SystemTime.getInstance().milliseconds() - startTimeInMs);
          if (failureCount.get() > maxFailuresPerPutBatchToStopPuts) {
            logger.error("{}  failures during this batch, hence exiting the Put Thread", failureCount.get());
            break;
          }
          Thread.sleep(threadLocalRandom.nextLong(sleepTimeBetweenBatchPutsInMs));
        }
      } catch (InterruptedException ex) {
        logger.error("InterruptedException in putThread ", ex);
      } finally {
        putComplete.set(true);
        metricsCollector.reportMetrics();
      }
      logger.info("Exiting Producer. Current Put Count {}, total Put Count {}", currentPutCount.get(), totalPutCount);
    }
Put threads to assist in producing blobs or uploading blobs to ambry
Put threads proceeds in multiple phases until {@code totalPutCount} is reached.
During each phase, N random uploads(max of {@code maxParallelRequest} are done to ambry with the help of
{@link PutGetHelper}. On completion of one phase, the thread proceeds onto the next phase after sleeping for max of
{@code sleepTimeBetweenBatchPutsInMs} ms.
ambry-tools/src/main/java/com/github/ambry/tools/admin/ConcurrencyTestTool.java