[BUG] Storage blobClient.openSeekableByteChannelRead makes unnecessary request to determine if stream has ended

[lmolkova]

Stress-test findings:

the code blobClient.openSeekableByteChannelRead(new BlobSeekableByteChannelReadOptions(), span)

results in the following HTTP requests

The first one does not know the content size and request the 4MB chunk, but all the content is returned in the first request and the second one (resulting in 416) is not necessary and can be optimized away.

Based on conversation with @ibrahimrabab and @alzimmermsft the SDK may do the following:

1. if ETag consistency is applied, we don't need to wait for 416 and can determine if we got everything based on the content-range response header for the first chunk.
2. If ETags are not used, then we should keep going until we get 416.
3. Also, currently, if an error happens when reading the body of 416 response, the openSeekableByteChannelRead throws even though the content is received fully.
   We should consider just logging a warning when this happens since we already know we got all the content and error has happened after.

Exception or Stack Trace

This exception is thrown when response body streaming fail when receiving 416 response (corresponding to the problem p3 above)

reactor.core.Exceptions$ReactiveException: io.netty.channel.unix.Errors$NativeIoException: recvAddress(..) failed: Connection reset by peer
	at reactor.core.Exceptions.propagate(Exceptions.java:396)
	at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:98)
	at reactor.core.publisher.Mono.block(Mono.java:1742)
	at com.azure.storage.common.implementation.StorageImplUtils.blockWithOptionalTimeout(StorageImplUtils.java:147)
	at com.azure.storage.blob.specialized.BlobClientBase.downloadStreamWithResponse(BlobClientBase.java:964)
	at com.azure.storage.blob.specialized.StorageSeekableByteChannelBlobReadBehavior.read(StorageSeekableByteChannelBlobReadBehavior.java:76)
	at com.azure.storage.common.implementation.StorageSeekableByteChannel.refillReadBuffer(StorageSeekableByteChannel.java:188)
	at com.azure.storage.common.implementation.StorageSeekableByteChannel.read(StorageSeekableByteChannel.java:166)
	at java.base/sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:65)
	at java.base/sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:107)
	at java.base/sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:101)
	at com.azure.storage.stress.CrcInputStream.read(CrcInputStream.java:60)
	at java.base/java.io.InputStream.read(InputStream.java:205)
	at com.azure.storage.blob.stress.OpenSeekableByteChannelRead.runInternal(OpenSeekableByteChannelRead.java:43)
	at com.azure.storage.blob.stress.BlobScenarioBase.run(BlobScenarioBase.java:84)
	at com.azure.perf.test.core.PerfStressTest.runTest(PerfStressTest.java:31)
	at com.azure.perf.test.core.ApiPerfTestBase.runAll(ApiPerfTestBase.java:156)
	at com.azure.perf.test.core.PerfStressProgram.lambda$runTests$12(PerfStressProgram.java:244)
	at java.base/java.util.concurrent.ForkJoinTask$AdaptedCallable.exec(ForkJoinTask.java:1448)
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)
	Suppressed: java.lang.Exception: #block terminated with an error
		at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:100)
		... 22 more
Caused by: io.netty.channel.unix.Errors$NativeIoException: recvAddress(..) failed: Connection reset by peer

To Reproduce
Reproduces with the following stress test:

azure-sdk-for-java/sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/OpenSeekableByteChannelRead.java

Lines 38 to 48 in abadbdf

	try {
	BlobSeekableByteChannelReadResult result = syncClient.openSeekableByteChannelRead(
	new BlobSeekableByteChannelReadOptions(), null);
	SeekableByteChannel channel = result.getChannel();
	return ORIGINAL_CONTENT.checkMatch(BinaryData.fromStream(Channels.newInputStream(channel)), span).block().booleanValue();
	} catch (Exception e) {
	LOGGER.error("Failed to download blob with open seekable byte channel", e);
	return false;
	}
	}

[seanmcc-msft]

We should fix this in the future.