class S3FileInput implements FileInput {

    private static final Pattern HAS_GLOBS_PATTERN = Pattern.compile("^((s3://)[^\\*]*/)[^\\*]*\\*.*");

    private AmazonS3 client; // to prevent early GC during getObjectContent() in getStream()
    private static final Logger LOGGER = LogManager.getLogger(S3FileInput.class);

    private final S3ClientHelper clientBuilder;

    @NotNull
    private final S3URI normalizedS3URI;
    @Nullable
    private final S3URI preGlobUri;
    @NotNull
    private final Predicate<S3URI> uriPredicate;
    @Nullable
    private final String protocolSetting;

    public S3FileInput(URI uri, String protocol) {
        this.clientBuilder = new S3ClientHelper();
        this.normalizedS3URI = S3URI.toS3URI(uri);
        this.preGlobUri = toPreGlobUri(this.normalizedS3URI);
        this.uriPredicate = new GlobPredicate(this.normalizedS3URI);
        this.protocolSetting = protocol;
    }

    @VisibleForTesting
    S3FileInput(S3ClientHelper clientBuilder, URI uri, String protocol) {
        this.clientBuilder = clientBuilder;
        this.normalizedS3URI = S3URI.toS3URI(uri);
        this.preGlobUri = toPreGlobUri(this.normalizedS3URI);
        this.uriPredicate = new GlobPredicate(this.normalizedS3URI);
        this.protocolSetting = protocol;
    }

    @Override
    public boolean isGlobbed() {
        return preGlobUri != null;
    }

    @Override
    public URI uri() {
        return normalizedS3URI.uri();
    }

    @Override
    public List<URI> expandUri() throws IOException {
        if (isGlobbed() == false) {
            return List.of(normalizedS3URI.uri());
        }
        if (client == null) {
            client = clientBuilder.client(preGlobUri, protocolSetting);
        }
        List<URI> uris = new ArrayList<>();
        ObjectListing list = client.listObjects(preGlobUri.bucket(), preGlobUri.key());
        addKeyUris(uris, list);
        while (list.isTruncated()) {
            list = client.listNextBatchOfObjects(list);
            addKeyUris(uris, list);
        }
        return uris;
    }

    private void addKeyUris(List<URI> uris, ObjectListing list) {
        assert preGlobUri != null;
        List<S3ObjectSummary> summaries = list.getObjectSummaries();
        for (S3ObjectSummary summary : summaries) {
            String key = summary.getKey();
            if (!key.endsWith("/")) {
                S3URI keyUri = preGlobUri.replacePath(summary.getBucketName(), key);
                if (uriPredicate.test(keyUri)) {
                    uris.add(keyUri.uri());
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("{}", keyUri);
                    }
                }
            }
        }
    }

    @Override
    public InputStream getStream(URI uri) throws IOException {
        S3URI s3URI = S3URI.toS3URI(uri);
        if (client == null) {
            client = clientBuilder.client(s3URI, protocolSetting);
        }
        S3Object object = client.getObject(s3URI.bucket(), s3URI.key());
        if (object != null) {
            return object.getObjectContent();
        }
        throw new IOException("Failed to load S3 URI: " + uri.toString());
    }

    @Override
    public boolean sharedStorageDefault() {
        return true;
    }

    @VisibleForTesting
    @Nullable
    static S3URI toPreGlobUri(S3URI uri) {<FILL_FUNCTION_BODY>}

    private static class GlobPredicate implements Predicate<S3URI> {
        private final Pattern globPattern;

        GlobPredicate(S3URI uri) {
            this.globPattern = Pattern.compile(Globs.toUnixRegexPattern(uri.toString()));
        }

        @Override
        public boolean test(@Nullable S3URI input) {
            return input != null && globPattern.matcher(input.toString()).matches();
        }
    }
}

        Matcher hasGlobMatcher = HAS_GLOBS_PATTERN.matcher(uri.toString());
        S3URI preGlobUri = null;
        if (hasGlobMatcher.matches()) {
            preGlobUri = S3URI.toS3URI(URI.create(hasGlobMatcher.group(1)));
        }
        return preGlobUri;
    
<no_super_class>