@SuppressWarnings("ChainOfInstanceofChecks")
  public OpenFileInformation prepareToOpenFile(
      final Path path,
      final OpenFileParameters parameters,
      final long blockSize) throws IOException {
    Configuration options = parameters.getOptions();
    Set<String> mandatoryKeys = parameters.getMandatoryKeys();
    // S3 Select is not supported in this release
    if (options.get(SelectConstants.SELECT_SQL, null) != null) {
      if (mandatoryKeys.contains(SelectConstants.SELECT_SQL)) {
        // mandatory option: fail with a specific message.
        throw new UnsupportedOperationException(SelectConstants.SELECT_UNSUPPORTED);
      } else {
        // optional; log once and continue
        LOG_NO_SQL_SELECT.warn(SelectConstants.SELECT_UNSUPPORTED);
      }
    }
    // choice of keys depends on open type
    rejectUnknownMandatoryKeys(
        mandatoryKeys,
        InternalConstants.S3A_OPENFILE_KEYS,
        "for " + path + " in file I/O");

    // where does a read end?
    long fileLength = LENGTH_UNKNOWN;

    // was a status passed in via a withStatus() invocation in
    // the builder API?
    FileStatus providedStatus = parameters.getStatus();
    S3AFileStatus fileStatus = null;
    if (providedStatus != null) {
      // there's a file status

      // make sure the file name matches -the rest of the path
      // MUST NOT be checked.
      Path providedStatusPath = providedStatus.getPath();
      checkArgument(path.getName().equals(providedStatusPath.getName()),
          "Filename mismatch between file being opened %s and"
              + " supplied filestatus %s",
          path, providedStatusPath);

      // make sure the status references a file
      if (providedStatus.isDirectory()) {
        throw new FileNotFoundException(
            "Supplied status references a directory " + providedStatus);
      }
      // build up the values
      long len = providedStatus.getLen();
      long modTime = providedStatus.getModificationTime();
      String versionId;
      String eTag;
      // can use this status to skip our own probes,
      LOG.debug("File was opened with a supplied FileStatus;"
              + " skipping getFileStatus call in open() operation: {}",
          providedStatus);

      // what type is the status (and hence: what information does it contain?)
      if (providedStatus instanceof S3AFileStatus) {
        // is it an S3AFileSystem status?
        S3AFileStatus st = (S3AFileStatus) providedStatus;
        versionId = st.getVersionId();
        eTag = st.getEtag();
      } else if (providedStatus instanceof S3ALocatedFileStatus) {

        //  S3ALocatedFileStatus instance may supply etag and version.
        S3ALocatedFileStatus st = (S3ALocatedFileStatus) providedStatus;
        versionId = st.getVersionId();
        eTag = st.getEtag();
      } else {
        // it is another type.
        // build a status struct without etag or version.
        LOG.debug("Converting file status {}", providedStatus);
        versionId = null;
        eTag = null;
      }
      // Construct a new file status with the real path of the file.
      fileStatus = new S3AFileStatus(
          len,
          modTime,
          path,
          blockSize,
          username,
          eTag,
          versionId);
      // set the end of the read to the file length
      fileLength = fileStatus.getLen();
    }
    FSBuilderSupport builderSupport = new FSBuilderSupport(options);
    // determine start and end of file.
    long splitStart = builderSupport.getPositiveLong(FS_OPTION_OPENFILE_SPLIT_START, 0);

    // split end
    long splitEnd = builderSupport.getLong(
        FS_OPTION_OPENFILE_SPLIT_END, LENGTH_UNKNOWN);

    if (splitStart > 0 && splitStart > splitEnd) {
      LOG.warn("Split start {} is greater than split end {}, resetting",
          splitStart, splitEnd);
      splitStart = 0;
    }

    // read end is the open file value
    fileLength = builderSupport.getPositiveLong(FS_OPTION_OPENFILE_LENGTH, fileLength);

    // if the read end has come from options, use that
    // in creating a file status
    if (fileLength >= 0 && fileStatus == null) {
      fileStatus = createStatus(path, fileLength, blockSize);
    }

    // Build up the input policy.
    // seek policy from default, s3a opt or standard option
    // read from the FS standard option.
    Collection<String> policies =
        options.getStringCollection(FS_OPTION_OPENFILE_READ_POLICY);
    if (policies.isEmpty()) {
      // fall back to looking at the S3A-specific option.
      policies = options.getStringCollection(INPUT_FADVISE);
    }

    return new OpenFileInformation()
        .withAsyncDrainThreshold(
            builderSupport.getPositiveLong(ASYNC_DRAIN_THRESHOLD,
                defaultReadAhead))
        .withBufferSize(
            (int)builderSupport.getPositiveLong(
                FS_OPTION_OPENFILE_BUFFER_SIZE, defaultBufferSize))
        .withChangePolicy(changePolicy)
        .withFileLength(fileLength)
        .withInputPolicy(
            S3AInputPolicy.getFirstSupportedPolicy(policies, defaultInputPolicy))
        .withReadAheadRange(
            builderSupport.getPositiveLong(READAHEAD_RANGE, defaultReadAhead))
        .withSplitStart(splitStart)
        .withSplitEnd(splitEnd)
        .withStatus(fileStatus)
        .build();

  }
@Test
  public void testUnknownMandatoryOption() throws Throwable {

    String key = "unknown";
    intercept(IllegalArgumentException.class, key, () ->
        prepareToOpenFile(params(key, "undefined")));
  }