@Override
        public CompletableFuture<Optional<StreamOffsetRange>> find(long objectId, long streamId) {
            S3Object s3Object = objectsImage.getObjectMetadata(objectId);
            if (s3Object == null) {
                return FutureUtil.failedFuture(new IllegalArgumentException("Cannot find object metadata for object: " + objectId));
            }
            // The reader will be release after the find operation
            @SuppressWarnings("resource")
            ObjectReader reader = objectReaderFactory.get(new S3ObjectMetadata(objectId, s3Object.getObjectSize(), s3Object.getAttributes()));
            CompletableFuture<Optional<StreamOffsetRange>> cf = reader.basicObjectInfo().thenApply(info -> info.indexBlock().findStreamOffsetRange(streamId));
            cf.whenComplete((rst, ex) -> reader.release());
            return cf;
        }
After use, the caller must call {@link Image#close()} to release the image.
core/src/main/scala/kafka/log/stream/s3/metadata/StreamMetadataManager.java