code: @Override
    public void commitBlock(long workerId, long usedBytesOnTier, String tierAlias, long blockId,
        long length) throws NotFoundException, UnavailableException {
      LOG.debug("Commit block from workerId: {}, usedBytesOnTier: {}, blockId: {}, length: {}",
          workerId, usedBytesOnTier, blockId, length);
  
      MasterWorkerInfo worker = mWorkers.getFirstByField(ID_INDEX, workerId);
      // TODO(peis): Check lost workers as well.
      if (worker == null) {
        throw new NotFoundException(ExceptionMessage.NO_WORKER_FOUND.getMessage(workerId));
      }
  
      // Lock the worker metadata first.
      try (JournalContext journalContext = createJournalContext()) {
        synchronized (worker) {
          try (LockResource lr = lockBlock(blockId)) {
            Optional<BlockMeta> block = mBlockStore.getBlock(blockId);
            if (!block.isPresent() || block.get().getLength() != length) {
              if (block.isPresent() && block.get().getLength() != Constants.UNKNOWN_SIZE) {
                LOG.warn("Rejecting attempt to change block length from {} to {}",
                    block.get().getLength(), length);
              } else {
                mBlockStore.putBlock(blockId, BlockMeta.newBuilder().setLength(length).build());
                BlockInfoEntry blockInfo =
                    BlockInfoEntry.newBuilder().setBlockId(blockId).setLength(length).build();
                journalContext.append(JournalEntry.newBuilder().setBlockInfo(blockInfo).build());
              }
            }
  
            // Update the block metadata with the new worker location.
            mBlockStore.addLocation(blockId, BlockLocation.newBuilder()
                .setWorkerId(workerId)
                .setTier(tierAlias)
                .build());
            // This worker has this block, so it is no longer lost.
            mLostBlocks.remove(blockId);
  
            // Update the worker information for this new block.
            // TODO(binfan): when retry commitBlock on master is expected, make sure metrics are not
            // double counted.
            worker.addBlock(blockId);
            worker.updateUsedBytes(tierAlias, usedBytesOnTier);
            worker.updateLastUpdatedTimeMs();
          }
        }
      }
    }
docs: TODO(binfan): check the logic is correct or not when commitBlock is a retry