Fix AzureSeekableByteChannel bugs related to buffer/channel boundaries (Azure#24723)

* AzureSeekableByteChannel - fix off-by-one & return -1 (EOF) when position() >= size()

* AzureSeekableByteChannel - read() must respect ByteBuffer.position() not assume 0

* AzureSeekableByteChannel - write() must respect ByteBuffer.position() and .limit()

* AzureSeekableByteChannel - fix test Groovy code compile failure on Java 8

* AzureSeekableByteChannel - attempt to fix Groovy test MissingMethodException

Error reported only by Azure Pipelines `windows2019_117_surefiretest` build, was:

[ERROR] Read respect dest buffer pos  Time elapsed: 0.019 s  <<< ERROR!
groovy.lang.MissingMethodException: 
No signature of method: [B.clone() is applicable for argument types: () values: []
Possible solutions: count(java.lang.Object), collect(), collect(groovy.lang.Closure), collect(java.util.Collection, groovy.lang.Closure), toSet(), size()
    at com.azure.storage.blob.nio.AzureSeekableByteChannelTest.Read respect dest buffer pos(AzureSeekableByteChannelTest.groovy:101)

Line 101 was: 
  def destArray = randArray.clone()

Other build checks using other platforms/versions pass, and cannot reproduce locally.
Changed to an explicit `new` followed by an `arraycopy`.

* AzureSeekableByteChannel - Added comments for requested clarification

Author: prosprice

sdk/storage/azure-storage-blob-nio/pom.xml

@@ -131,6 +131,12 @@
       <version>1.7.4</version> <!-- {x-version-update;com.azure:azure-core-http-okhttp;dependency} -->
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <version>3.12.4</version> <!-- {x-version-update;org.mockito:mockito-core;external_dependency} -->
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

sdk/storage/azure-storage-blob-nio/src/main/java/com/azure/storage/blob/nio/AzureSeekableByteChannel.java

@@ -63,42 +63,49 @@ public int read(ByteBuffer dst) throws IOException {
         validateOpen();
         validateReadMode();
 
-        // See comments in position()
-        if (this.position > this.size()) {
-            return -1;
+        // See comments in position(), remember that position is 0-based and size() is exclusive
+        if (this.position >= this.size()) {
+            return -1; // at or past EOF
         }
 
-        int count = 0;
-
-        int len = dst.remaining();
-        byte[] buf;
-
         // If the buffer is backed by an array, we can write directly to that instead of allocating new memory.
+        int pos;
+        final int limit;
+        final byte[] buf;
         if (dst.hasArray()) {
+            // ByteBuffer has a position and limit that define the bounds of the writeable area, and that
+            // area can be both smaller than the backing array and might not begin at array index 0.
+            pos = dst.position();
+            limit = pos + dst.remaining();
             buf = dst.array();
         } else {
-            buf = new byte[len];
+            pos = 0;
+            limit = dst.remaining();
+            buf = new byte[limit];
         }
 
-        while (count < len) {
-            int retCount = this.reader.read(buf, count, len - count);
-            if (retCount == -1) {
+        while (pos < limit) {
+            int byteCount = this.reader.read(buf, pos, limit - pos);
+            if (byteCount == -1) {
                 break;
             }
-            count += retCount;
+            pos += byteCount;
         }
 
         /*
         Either write to the destination if we had to buffer separately or just set the position correctly if we wrote
         underneath the buffer
          */
-        if (!dst.hasArray()) {
-            dst.put(buf, 0, count);
+        int count;
+        if (dst.hasArray()) {
+            count = pos - dst.position();
+            dst.position(pos);
         } else {
-            dst.position(dst.position() + count); //
+            count = pos; // original position was 0
+            dst.put(buf, 0, count);
         }
-        this.position += count;
 
+        this.position += count;
         return count;
     }
 
@@ -108,25 +115,30 @@ public int write(ByteBuffer src) throws IOException {
         validateOpen();
         validateWriteMode();
 
-        int length = src.remaining();
-
-        this.position += src.remaining();
+        final int length = src.remaining();
+        this.position += length;
 
         /*
         If the buffer is backed by an array, we can read directly from that instead of allocating new memory.
-
         Set the position correctly if we read from underneath the buffer
          */
+        int pos;
         byte[] buf;
         if (src.hasArray()) {
+            // ByteBuffer has a position and limit that define the bounds of the readable area, and that
+            // area can be both smaller than the backing array and might not begin at array index 0.
+            pos = src.position();
             buf = src.array();
-            src.position(src.position() + length);
+            src.position(pos + length);
         } else {
+            pos = 0;
             buf = new byte[length];
-            src.get(buf);
+            src.get(buf); // advances src.position()
         }
-        this.writer.write(buf);
-
+        // Either way, the src.position() and this.position have been updated before we know if this write
+        // will succeed. (Original behavior.) It may be better to update position(s) only *after* success,
+        // but then on IOException would we know if there was a partial write, and if so how much?
+        this.writer.write(buf, pos, length);
         return length;
     }
 
@@ -209,6 +221,10 @@ public void close() throws IOException {
         this.closed = true;
     }
 
+    Path getPath() {
+        return this.path;
+    }
+
     private void validateOpen() throws ClosedChannelException {
         if (this.closed) {
             throw LoggingUtility.logError(logger, new ClosedChannelException());

sdk/storage/azure-storage-blob-nio/src/test/java/com/azure/storage/blob/nio/AzureSeekableByteChannelTest.groovy

@@ -4,6 +4,9 @@
 package com.azure.storage.blob.nio;
 
 import com.azure.storage.blob.BlobClient
+import com.azure.storage.blob.specialized.BlobOutputStream
+import org.mockito.Answers
+import org.mockito.Mockito
 import spock.lang.Unroll
 
 import java.nio.ByteBuffer
@@ -63,6 +66,68 @@ class AzureSeekableByteChannelTest extends APISpec {
         os.toByteArray() == fileContent
     }
 
+    def "Read loop until EOF"() {
+        setup:
+        def fileContent = new byte[sourceFileSize]
+        fileStream.read(fileContent)
+        def os = new ByteArrayOutputStream(sourceFileSize)
+        def rand = new Random()
+        long timeLimit = System.currentTimeMillis() + 60_000 // fail if test runs >= 1 minute
+
+        when:
+        while (System.currentTimeMillis() < timeLimit) { // ensures test duration is bounded
+            def buffer = ByteBuffer.allocate(rand.nextInt(1024 * 1024))
+            int readAmount = readByteChannel.read(buffer)
+            if (readAmount == -1) {
+                break; // reached EOF
+            }
+            os.write(buffer.array(), 0, readAmount) // limit the write in case we allocated more than we needed
+        }
+
+        then:
+        os.toByteArray() == fileContent
+        System.currentTimeMillis() < timeLimit // else potential inf. loop if read() always returns 0
+    }
+
+    def "Read respect dest buffer pos"() {
+        setup:
+        def fileContent = new byte[sourceFileSize]
+        fileStream.read(fileContent)
+
+        def rand = new Random()
+        int initialOffset = rand.nextInt(512) + 1 // always > 0
+        byte[] randArray = new byte[2 * initialOffset + sourceFileSize]
+        rand.nextBytes(randArray) // fill with random bytes
+
+        // copy same random bytes, but in this copy some will eventually be overwritten by read()
+        byte[] destArray = new byte[randArray.length]
+        System.arraycopy(randArray, 0, destArray, 0, randArray.length)
+        def dest = ByteBuffer.wrap(destArray)
+        dest.position(initialOffset) // will have capacity on either side that should not be touched
+
+        when:
+        int readAmount = 0;
+        while (readAmount != -1) {
+            assert dest.position() != 0
+            readAmount = readByteChannel.read(dest) // backed by an array, but position != 0
+        }
+
+        then:
+        dest.position() == initialOffset + sourceFileSize
+        compareInputStreams( // destination content should match file content at initial read position
+            new ByteArrayInputStream(destArray, initialOffset, sourceFileSize),
+            new ByteArrayInputStream(fileContent),
+            sourceFileSize)
+        compareInputStreams( // destination content should be untouched prior to initial position
+            new ByteArrayInputStream(destArray, 0, initialOffset),
+            new ByteArrayInputStream(randArray, 0, initialOffset),
+            initialOffset)
+        compareInputStreams( // destination content should be untouched past end of read
+            new ByteArrayInputStream(destArray, initialOffset + sourceFileSize, initialOffset),
+            new ByteArrayInputStream(randArray, initialOffset + sourceFileSize, initialOffset),
+            initialOffset)
+    }
+
     def "Read fs close"() {
         when:
         fs.close()
@@ -96,6 +161,61 @@ class AzureSeekableByteChannelTest extends APISpec {
         compareInputStreams(writeBc.openInputStream(), new ByteArrayInputStream(fileContent), sourceFileSize)
     }
 
+    def "Write respect src buffer pos"() {
+        setup:
+        def rand = new Random()
+        int initialOffset = rand.nextInt(512) + 1 // always > 0
+        def srcBufferContent = new byte[2 * initialOffset + sourceFileSize]
+        rand.nextBytes(srcBufferContent) // fill with random bytes
+
+        def fileContent = new byte[sourceFileSize]
+        fileStream.read(fileContent)
+
+        // place expected file content into source buffer at random location, retain other random bytes
+        System.arraycopy(fileContent, 0, srcBufferContent, initialOffset, sourceFileSize)
+        def srcBuffer = ByteBuffer.wrap(srcBufferContent)
+        srcBuffer.position(initialOffset)
+        srcBuffer.limit(initialOffset + sourceFileSize)
+
+        // This test aims to observe the actual bytes written by the ByteChannel to the underlying OutputStream,
+        // not just the number of bytes allegedly written as reported by its position. It would prefer to examine
+        // the OutputStream directly, but the channel requires the specific NioBlobOutputStream implementation
+        // and does not accept something generic like a ByteArrayOutputStream. NioBlobOutputStream is final, so
+        // it cannot be subclassed or mocked and has little state of its own -- writes go to a BlobOutputStream.
+        // That class is abstract, but its constructor is not accessible outside its package and cannot normally
+        // be subclassed to provide custom behavior, but a runtime mocking framework like Mockito can. This is
+        // the nearest accessible observation point, so the test mocks a BlobOutputStream such that all write
+        // methods store data in ByteArrayOutputStream which it can later examine for its size and content.
+        def actualOutput = new ByteArrayOutputStream(sourceFileSize)
+        def blobOutputStream = Mockito.mock(
+            BlobOutputStream.class, Mockito.withSettings().useConstructor(4096 /* block size */))
+        Mockito.doAnswer( { invoked -> actualOutput.write(invoked.getArgument(0)) } )
+            .when(blobOutputStream).write(Mockito.anyInt())
+        Mockito.doAnswer( { invoked -> actualOutput.writeBytes(invoked.getArgument(0)) } )
+            .when(blobOutputStream).write(Mockito.any(byte[].class))
+        Mockito.doAnswer( { invoked -> actualOutput.write(
+                invoked.getArgument(0), invoked.getArgument(1), invoked.getArgument(2)) } )
+            .when(blobOutputStream).write(Mockito.any(byte[].class), Mockito.anyInt(), Mockito.anyInt())
+        def path = writeByteChannel.getPath()
+        writeByteChannel = new AzureSeekableByteChannel(new NioBlobOutputStream(blobOutputStream, path), path)
+
+        when:
+        int written = 0
+        while (written < sourceFileSize) {
+            written += writeByteChannel.write(srcBuffer)
+        }
+        writeByteChannel.close()
+
+        then:
+        srcBuffer.position() == initialOffset + sourceFileSize // src buffer position SHOULD be updated
+        srcBuffer.limit() == srcBuffer.position() // limit SHOULD be unchanged (still at end of content)
+        // the above report back to the caller, but this verifies the correct bytes are going to the blob:
+        compareInputStreams(
+            new ByteArrayInputStream(actualOutput.toByteArray()),
+            new ByteArrayInputStream(fileContent),
+            sourceFileSize)
+    }
+
     def "Write fs close"() {
         when:
         fs.close()
@@ -219,10 +339,10 @@ class AzureSeekableByteChannelTest extends APISpec {
         thrown(IllegalArgumentException)
 
         when:
-        readByteChannel.position(sourceFileSize + 1)
+        readByteChannel.position(sourceFileSize) // position is 0-based, so seeking to size --> EOF
 
         then:
-        readByteChannel.read(ByteBuffer.allocate(1)) == -1 // Seeking past the end and then reading should indicate EOF
+        readByteChannel.read(ByteBuffer.allocate(1)) == -1 // Seeking to the end and then reading should indicate EOF
     }
 
     def "Seek fs close"() {