public InputStream acquire(UnderFileSystem ufs, String path, FileId fileId,
      OpenOptions openOptions) throws IOException {
    if (!ufs.isSeekable() || !CACHE_ENABLED) {
      // only seekable UFSes are cachable/reusable, always return a new input stream
      return ufs.openExistingFile(path, openOptions);
    }

    // explicit cache cleanup
    try {
      mStreamCache.cleanUp();
    } catch (Throwable error) {
      SAMPLING_LOG.warn("Explicit cache removal failed.", error);
    }

    StreamIdSet streamIds = mFileIdToStreamIds.compute(fileId, (key, value) -> {
      if (value != null) {
        return value;
      }
      return new StreamIdSet();
    });

    // Try to acquire an existing id from the stream id set.
    // synchronized is required to be consistent between availableIds() and acquire(id).
    CachedSeekableInputStream inputStream = null;
    synchronized (streamIds) {
      // find the next available input stream from the cache
      for (long id : streamIds.availableIds()) {
        inputStream = mStreamCache.getIfPresent(id);
        if (inputStream != null) {
          // acquire it now while locked, so other threads cannot take it
          streamIds.acquire(id);
          break;
        }
      }
    }

    if (inputStream != null) {
      // for the cached ufs instream, seek (outside of critical section) to the requested position.
      LOG.debug("Reused the under file input stream resource of {}", inputStream.getResourceId());
      inputStream.seek(openOptions.getOffset());
      return inputStream;
    }

    // no cached input stream is available, acquire a new id and open a new stream
    final long newId = streamIds.acquireNewId();
    try {
      inputStream = mStreamCache.get(newId, () -> {
        SeekableUnderFileInputStream ufsStream = (SeekableUnderFileInputStream) ufs
            .openExistingFile(path,
                OpenOptions.defaults().setPositionShort(openOptions.getPositionShort())
                    .setOffset(openOptions.getOffset()));
        LOG.debug("Created the under file input stream resource of {}", newId);
        return new CachedSeekableInputStream(ufsStream, newId, fileId, path);
      });
    } catch (ExecutionException e) {
      LOG.warn("Failed to create a new cached ufs instream of file id {} and path {}", fileId,
          path, e);
      // fall back to an uncached ufs creation.
      return ufs.openExistingFile(path,
          OpenOptions.defaults().setOffset(openOptions.getOffset()));
    }
    catch (UncheckedExecutionException e) {
      throw AlluxioRuntimeException.from(e.getCause());
    }
    return inputStream;
  }
@Test
  public void multipleCheckIn() throws Exception {
    mManager.acquire(mUfs, FILE_NAME, FILE_ID, OpenOptions.defaults().setOffset(2));
    mManager.acquire(mUfs, FILE_NAME, FILE_ID, OpenOptions.defaults().setOffset(4));
    mManager.acquire(mUfs, FILE_NAME, FILE_ID, OpenOptions.defaults().setOffset(6));
    // 3 different input streams are acquired
    verify(mUfs, times(3)).openExistingFile(eq(FILE_NAME),
        any(OpenOptions.class));
  }