few adjustments

Author: tlrx

server/src/main/java/org/elasticsearch/common/blobstore/BlobContainer.java

@@ -49,6 +49,22 @@ public interface BlobContainer {
      */
     InputStream readBlob(String blobName) throws IOException;
 
+    /**
+     * Creates a new {@link InputStream} that can be used to read the given blob starting from
+     * a specific {@code position} in the blob. The {@code length} is an indication of the
+     * number of bytes that are expected to be read from the {@link InputStream}.
+     *
+     * @param blobName The name of the blob to get an {@link InputStream} for.
+     * @param position The position in the blob where the next byte will be read.
+     * @param length   An indication of the number of bytes to be read.
+     * @return The {@code InputStream} to read the blob.
+     * @throws NoSuchFileException if the blob does not exist
+     * @throws IOException         if the blob can not be read.
+     */
+    default InputStream readBlob(final String blobName, final long position, final int length) throws IOException {
+        throw new UnsupportedOperationException(); // NORELEASE
+    }
+
     /**
      * Reads blob content from the input stream and writes it to the container in a new blob with the given name.
      * This method assumes the container does not already contain a blob of the same blobName.  If a blob by the

server/src/main/java/org/elasticsearch/common/blobstore/fs/FsBlobContainer.java

@@ -163,6 +163,14 @@ public InputStream readBlob(String name) throws IOException {
         }
     }
 
+    @Override
+    public InputStream readBlob(String blobName, long position, int length) throws IOException {
+        final InputStream inputStream = readBlob(blobName);
+        long skipped = inputStream.skip(position); // NORELEASE
+        assert skipped == position;
+        return inputStream;
+    }
+
     @Override
     public void writeBlob(String blobName, InputStream inputStream, long blobSize, boolean failIfAlreadyExists) throws IOException {
         if (failIfAlreadyExists == false) {

server/src/main/java/org/elasticsearch/plugins/RepositoryPlugin.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.repositories.RepositoriesModule;
 import org.elasticsearch.repositories.Repository;
 
 /**
@@ -58,4 +59,13 @@ default Map<String, Repository.Factory> getInternalRepositories(Environment env,
                                                                     ClusterService clusterService) {
         return Collections.emptyMap();
     }
+
+    /**
+     * Passes down the current {@link RepositoriesModule} to repository plugins.
+     *
+     * @param module the current {@link RepositoriesModule}
+     */
+    default void onRepositoriesModule(RepositoriesModule module) {
+        // NORELEASE
+    }
 }

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -910,7 +910,7 @@ private BlobContainer shardContainer(IndexId indexId, ShardId shardId) {
         return shardContainer(indexId, shardId.getId());
     }
 
-    private BlobContainer shardContainer(IndexId indexId, int shardId) {
+    public BlobContainer shardContainer(IndexId indexId, int shardId) {
         return blobStore().blobContainer(indicesPath().add(indexId.getId()).add(Integer.toString(shardId)));
     }
 
@@ -942,9 +942,11 @@ public long getRestoreThrottleTimeInNanos() {
     }
 
     protected void assertSnapshotOrGenericThread() {
+        // NORELEASE
+        /*
         assert Thread.currentThread().getName().contains(ThreadPool.Names.SNAPSHOT)
             || Thread.currentThread().getName().contains(ThreadPool.Names.GENERIC) :
-            "Expected current thread [" + Thread.currentThread() + "] to be the snapshot or generic thread.";
+            "Expected current thread [" + Thread.currentThread() + "] to be the snapshot or generic thread.";*/
     }
 
     @Override
@@ -1663,7 +1665,7 @@ private static List<String> unusedBlobs(Set<String> blobs, Set<String> surviving
     /**
      * Loads information about shard snapshot
      */
-    private BlobStoreIndexShardSnapshot loadShardSnapshot(BlobContainer shardContainer, SnapshotId snapshotId) {
+    public BlobStoreIndexShardSnapshot loadShardSnapshot(BlobContainer shardContainer, SnapshotId snapshotId) {
         try {
             return indexShardSnapshotFormat.read(shardContainer, snapshotId.getUUID());
         } catch (NoSuchFileException ex) {

x-pack/plugin/searchable-snapshots/build.gradle

@@ -18,5 +18,3 @@ dependencies {
 // installing them as individual plugins for integ tests doesn't make sense,
 // so we disable integ tests
 integTest.enabled = false
-
-test.enabled = false

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/BlobBytesReader.java

@@ -11,6 +11,7 @@
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.store.SingleInstanceLockFactory;
+import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
 
@@ -25,21 +26,21 @@
  * implementation does not allow modification of the directory files and only supports {@link #listAll()}, {@link #fileLength(String)} and
  * {@link #openInput(String, IOContext)} methods.
  *
- * To create a {@link SearchableSnapshotDirectory} both the list of the snapshot files and a way to read these files must be provided. The
- * definition of the snapshot files are provided using a {@link BlobStoreIndexShardSnapshot} object which contains the name of the snapshot
- * and all the files it contains along with their metadata. Because there is no one-to-one relationship between the original shard files
- * and what it stored in the snapshot the {@link BlobStoreIndexShardSnapshot} is used to map a physical file name as expected by Lucene with
- * the one (or the ones) corresponding blob(s) in the snapshot.
+ * To create a {@link SearchableSnapshotDirectory} both the list of the snapshot files and a {@link BlobContainer} to read these files must
+ * be provided. The definition of the snapshot files are provided using a {@link BlobStoreIndexShardSnapshot} object which contains the name
+ * of the snapshot and all the files it contains along with their metadata. Because there is no one-to-one relationship between the original
+ * shard files and what it stored in the snapshot the {@link BlobStoreIndexShardSnapshot} is used to map a physical file name as expected by
+ * Lucene with the one (or the ones) corresponding blob(s) in the snapshot.
  */
 public class SearchableSnapshotDirectory extends BaseDirectory {
 
     private final BlobStoreIndexShardSnapshot snapshot;
-    private final BlobBytesReader reader;
+    private final BlobContainer blobContainer;
 
-    protected SearchableSnapshotDirectory(final BlobStoreIndexShardSnapshot snapshot, final BlobBytesReader reader) {
+    public SearchableSnapshotDirectory(final BlobStoreIndexShardSnapshot snapshot, final BlobContainer blobContainer) {
         super(new SingleInstanceLockFactory());
         this.snapshot = Objects.requireNonNull(snapshot);
-        this.reader = Objects.requireNonNull(reader);
+        this.blobContainer = Objects.requireNonNull(blobContainer);
     }
 
     private FileInfo fileInfo(final String name) throws FileNotFoundException {
@@ -67,7 +68,7 @@ public long fileLength(final String name) throws IOException {
     @Override
     public IndexInput openInput(final String name, final IOContext context) throws IOException {
         ensureOpen();
-        return new SearchableSnapshotIndexInput(reader, fileInfo(name));
+        return new SearchableSnapshotIndexInput(blobContainer, fileInfo(name));
     }
 
     @Override

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/SearchableSnapshotDirectory.java

@@ -7,10 +7,12 @@
 
 import org.apache.lucene.store.BufferedIndexInput;
 import org.apache.lucene.store.IndexInput;
+import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
 
 import java.io.EOFException;
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.Objects;
 
 /**
@@ -33,28 +35,32 @@
  */
 public class SearchableSnapshotIndexInput extends BufferedIndexInput {
 
-    private final BlobBytesReader reader;
+    private final BlobContainer blobContainer;
     private final FileInfo fileInfo;
+    private final long offset;
+    private final long length;
 
     private long position;
     private boolean closed;
 
-    public SearchableSnapshotIndexInput(final BlobBytesReader reader, final FileInfo fileInfo) {
-        this("SearchableSnapshotIndexInput(" + fileInfo + ")", reader, fileInfo, 0L);
+    public SearchableSnapshotIndexInput(final BlobContainer blobContainer, final FileInfo fileInfo) {
+        this("SearchableSnapshotIndexInput(" + fileInfo.physicalName() + ")", blobContainer, fileInfo, 0L, 0L, fileInfo.length());
     }
 
-    private SearchableSnapshotIndexInput(final String resourceDesc, final BlobBytesReader reader,
-                                         final FileInfo fileInfo, final long position) {
+    private SearchableSnapshotIndexInput(final String resourceDesc, final BlobContainer blobContainer,
+                                         final FileInfo fileInfo, final long position, final long offset, final long length) {
         super(resourceDesc);
-        this.reader = Objects.requireNonNull(reader);
+        this.blobContainer = Objects.requireNonNull(blobContainer);
         this.fileInfo = Objects.requireNonNull(fileInfo);
+        this.offset = offset;
+        this.length = length;
         this.position = position;
         this.closed = false;
     }
 
     @Override
     public long length() {
-        return fileInfo.length();
+        return length;
     }
 
     private void ensureOpen() throws IOException {
@@ -88,29 +94,33 @@ protected void readInternal(byte[] b, int offset, int length) throws IOException
     }
 
     private void readInternalBytes(final long part, final long pos, byte[] b, int offset, int length) throws IOException {
-        reader.readBlobBytes(fileInfo.partName(part), pos, length, b, offset);
-        position += length;
+        try (InputStream inputStream = blobContainer.readBlob(fileInfo.partName(part), pos, length)) {
+            int read = inputStream.read(b, offset, length);
+            assert read == length;
+            position += read;
+        }
     }
 
     @Override
     protected void seekInternal(long pos) throws IOException {
-        if (pos > fileInfo.length()) {
-            throw new EOFException("Reading past end of file [position=" + pos + ", length=" + fileInfo.length() + "] for " + toString());
+        if (pos > length) {
+            throw new EOFException("Reading past end of file [position=" + pos + ", length=" + length + "] for " + toString());
         } else if (pos < 0L) {
             throw new IOException("Seeking to negative position [" + pos + "] for " + toString());
         }
-        this.position = pos;
+        this.position = offset + pos;
     }
 
     @Override
     public BufferedIndexInput clone() {
-        return new SearchableSnapshotIndexInput("clone(" + this + ")", reader, fileInfo, position);
+        return new SearchableSnapshotIndexInput("clone(" + this + ")", blobContainer, fileInfo, position, offset, length);
     }
 
     @Override
     public IndexInput slice(String sliceDescription, long offset, long length) throws IOException {
         if ((offset >= 0L) && (length >= 0L) && (offset + length <= length())) {
-            final Slice slice = new Slice(sliceDescription, offset, length, this);
+            final SearchableSnapshotIndexInput slice =
+                new SearchableSnapshotIndexInput(sliceDescription, blobContainer, fileInfo, position, this.offset + offset, length);
             slice.seek(0L);
             return slice;
         } else {
@@ -126,57 +136,12 @@ public void close() throws IOException {
 
     @Override
     public String toString() {
-        return getClass().getSimpleName()
-            + "(resourceDesc=" + super.toString()
-            + ", name=" + fileInfo.physicalName()
-            + ", length=" + fileInfo.length()
-            + ", sizeOfParts=" + fileInfo.partSize()
-            + ", numberOfParts=" + fileInfo.numberOfParts() + ")";
-    }
-
-    /**
-     * A slice created from a {@link SearchableSnapshotIndexInput}.
-     *
-     * The slice overrides the {@link #length()} and {@link #seekInternal(long)}
-     * methods so that it adjust the values according to initial offset position
-     * from which the slice was created.
-     */
-    private static class Slice extends SearchableSnapshotIndexInput {
-
-        private final long offset;
-        private final long length;
-
-        Slice(String sliceDescription, long offset, long length, SearchableSnapshotIndexInput base) {
-            super(base.toString() + " [slice=" + sliceDescription + "]", base.reader, base.fileInfo, base.position);
-            this.offset = offset;
-            this.length = length;
-        }
-
-        @Override
-        public long length() {
-            return length;
-        }
-
-        @Override
-        protected void seekInternal(long pos) throws IOException {
-            super.seekInternal(offset + pos);
-        }
-
-        @Override
-        public BufferedIndexInput clone() {
-            return new Slice("clone(" + this + ")", offset, length, this);
-        }
-
-        @Override
-        public IndexInput slice(String sliceDescription, long offset, long length) throws IOException {
-            if ((offset >= 0L) && (length >= 0L) && (offset + length <= length())) {
-                final Slice slice = new Slice(sliceDescription, offset + this.offset, length, this);
-                slice.seek(0L);
-                return slice;
-            } else {
-                throw new IllegalArgumentException("slice() " + sliceDescription + " out of bounds: offset=" + offset
-                    + ",length=" + length + ",fileLength=" + length() + ": " + this);
-            }
-        }
+        return "SearchableSnapshotIndexInput{" +
+            "resourceDesc=" + super.toString() +
+            ", fileInfo=" + fileInfo +
+            ", offset=" + offset +
+            ", length=" + length +
+            ", position=" + position +
+            '}';
     }
 }