@Override
  public boolean lock(Duration timeout) {
    //Start timer for timeout
    long startTime = System.currentTimeMillis();
    long lockTimeout = timeout.toMillis();
    Random random = new Random();

    while ((System.currentTimeMillis() - startTime) < lockTimeout) {
      try {
        leaseId.getAndSet(leaseBlobManager.acquireLease(LEASE_TIME_IN_SEC, leaseId.get()));
      } catch (AzureException e) {
        return false;
      }
      if (leaseId.get() != null) {
        LOG.info("Acquired lock!");
        hasLock.getAndSet(true);
        return true;
      } else {
        try {
          Thread.sleep(random.nextInt(1000));
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
        LOG.info("Trying to acquire lock again...");
      }
    }
    return false;
  }
Tries to acquire a lock in order to create intermediate streams. On failure to acquire lock, it keeps trying until the lock times out.
The lock is acquired when the blob is leased successfully.
@param timeout Duration after which timeout occurs.
@return true if the lock was acquired successfully, false if lock acquire operation is unsuccessful even after subsequent tries within the timeout range.
samza-azure/src/main/java/org/apache/samza/coordinator/AzureLock.java