Index: oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureSegmentArchiveReader.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureSegmentArchiveReader.java	(date 1537372353000)
+++ oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureSegmentArchiveReader.java	(date 1537199542000)
@@ -16,14 +16,9 @@
  */
 package org.apache.jackrabbit.oak.segment.azure;
 
-import com.google.common.base.Stopwatch;
-import com.microsoft.azure.storage.StorageException;
-import com.microsoft.azure.storage.blob.CloudBlob;
-import com.microsoft.azure.storage.blob.CloudBlobDirectory;
-import com.microsoft.azure.storage.blob.CloudBlockBlob;
-import org.apache.jackrabbit.oak.segment.spi.monitor.IOMonitor;
-import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveEntry;
-import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveReader;
+import static java.lang.Boolean.getBoolean;
+import static org.apache.jackrabbit.oak.segment.azure.AzureUtilities.getSegmentFileName;
+import static org.apache.jackrabbit.oak.segment.azure.AzureUtilities.readBufferFully;
 
 import java.io.File;
 import java.io.IOException;
@@ -37,10 +32,17 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
-import static org.apache.jackrabbit.oak.segment.azure.AzureUtilities.getSegmentFileName;
-import static org.apache.jackrabbit.oak.segment.azure.AzureUtilities.readBufferFully;
+import com.google.common.base.Stopwatch;
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudBlob;
+import com.microsoft.azure.storage.blob.CloudBlobDirectory;
+import com.microsoft.azure.storage.blob.CloudBlockBlob;
+import org.apache.jackrabbit.oak.segment.spi.monitor.IOMonitor;
+import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveEntry;
+import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveReader;
 
 public class AzureSegmentArchiveReader implements SegmentArchiveReader {
+    static final boolean OFF_HEAP = getBoolean("access.off.heap");
 
     private final CloudBlobDirectory archiveDirectory;
 
@@ -73,7 +75,13 @@
         if (indexEntry == null) {
             return null;
         }
-        ByteBuffer buffer = ByteBuffer.allocate(indexEntry.getLength());
+
+        ByteBuffer buffer;
+        if (OFF_HEAP) {
+            buffer = ByteBuffer.allocateDirect(indexEntry.getLength());
+        } else {
+            buffer = ByteBuffer.allocate(indexEntry.getLength());
+        }
         ioMonitor.beforeSegmentRead(pathAsFile(), msb, lsb, indexEntry.getLength());
         Stopwatch stopwatch = Stopwatch.createStarted();
         readBufferFully(getBlob(getSegmentFileName(indexEntry)), buffer);
Index: oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureSegmentArchiveWriter.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureSegmentArchiveWriter.java	(date 1537372353000)
+++ oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureSegmentArchiveWriter.java	(date 1537199542000)
@@ -16,6 +16,7 @@
  */
 package org.apache.jackrabbit.oak.segment.azure;
 
+import static org.apache.jackrabbit.oak.segment.azure.AzureSegmentArchiveReader.OFF_HEAP;
 import static org.apache.jackrabbit.oak.segment.azure.AzureUtilities.getSegmentFileName;
 import static org.apache.jackrabbit.oak.segment.azure.AzureUtilities.readBufferFully;
 
@@ -108,7 +109,13 @@
         if (indexEntry == null) {
             return null;
         }
-        ByteBuffer buffer = ByteBuffer.allocate(indexEntry.getLength());
+
+        ByteBuffer buffer;
+        if (OFF_HEAP) {
+            buffer = ByteBuffer.allocateDirect(indexEntry.getLength());
+        } else {
+            buffer = ByteBuffer.allocate(indexEntry.getLength());
+        }
         readBufferFully(getBlob(getSegmentFileName(indexEntry)), buffer);
         return buffer;
     }
Index: oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureUtilities.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureUtilities.java	(date 1537372353000)
+++ oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureUtilities.java	(date 1537199542000)
@@ -16,18 +16,8 @@
  */
 package org.apache.jackrabbit.oak.segment.azure;
 
-import com.microsoft.azure.storage.CloudStorageAccount;
-import com.microsoft.azure.storage.StorageCredentials;
-import com.microsoft.azure.storage.StorageException;
-import com.microsoft.azure.storage.StorageUri;
-import com.microsoft.azure.storage.blob.BlobListingDetails;
-import com.microsoft.azure.storage.blob.CloudBlob;
-import com.microsoft.azure.storage.blob.CloudBlobContainer;
-import com.microsoft.azure.storage.blob.CloudBlobDirectory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
+import java.io.OutputStream;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.ByteBuffer;
@@ -38,6 +28,18 @@
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
+import com.microsoft.azure.storage.CloudStorageAccount;
+import com.microsoft.azure.storage.StorageCredentials;
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.StorageUri;
+import com.microsoft.azure.storage.blob.BlobListingDetails;
+import com.microsoft.azure.storage.blob.CloudBlob;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import com.microsoft.azure.storage.blob.CloudBlobDirectory;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 public final class AzureUtilities {
 
     public static String SEGMENT_FILE_NAME_PATTERN = "^([0-9a-f]{4})\\.([0-9a-f-]+)$";
@@ -75,11 +77,8 @@
 
     public static void readBufferFully(CloudBlob blob, ByteBuffer buffer) throws IOException {
         try {
-            buffer.rewind();
-            long readBytes = blob.downloadToByteArray(buffer.array(), 0);
-            if (buffer.limit() != readBytes) {
-                throw new IOException("Buffer size: " + buffer.limit() + ", read bytes: " + readBytes);
-            }
+            blob.download(new ByteBufferOutputStream(buffer));
+            buffer.flip();
         } catch (StorageException e) {
             throw new IOException(e);
         }
@@ -112,4 +111,26 @@
 
         return container.getDirectoryReference(dir);
     }
+
+    private static class ByteBufferOutputStream extends OutputStream {
+
+        @NotNull
+        private final ByteBuffer buffer;
+
+        public ByteBufferOutputStream(@NotNull ByteBuffer buffer) {
+            this.buffer = buffer;
+        }
+
+        @Override
+        public void write(int b) throws IOException {
+            buffer.put((byte)b);
+        }
+
+        @Override
+        public void write(@NotNull byte[] bytes, int offset, int length) throws IOException {
+            buffer.put(bytes, offset, length);
+        }
+    }
 }
+
+