This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 620b8c55bd fix azure blob storage read position tracking, fixes #8838
(#8839)
620b8c55bd is described below
commit 620b8c55bdd963c0503641b2dfbd7444843c030f
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Sat Oct 10 11:31:27 2026 +0200
fix azure blob storage read position tracking, fixes #8838 (#8839)
---
.../org/apache/hop/vfs/azure/BlobInputStream.java | 87 +++++++-----
.../hop/vfs/azure/BlobInputStreamPositionTest.java | 158 +++++++++++++++++++++
2 files changed, 208 insertions(+), 37 deletions(-)
diff --git
a/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/BlobInputStream.java
b/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/BlobInputStream.java
index 5eef08bf1f..252d0761fd 100644
---
a/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/BlobInputStream.java
+++
b/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/BlobInputStream.java
@@ -19,73 +19,84 @@
package org.apache.hop.vfs.azure;
import
com.azure.storage.file.datalake.models.DataLakeFileOpenInputStreamResult;
+import com.azure.storage.file.datalake.models.PathProperties;
import java.io.IOException;
import java.io.InputStream;
+/**
+ * Reads an Azure file and never returns more than {@code fileSize} bytes: a
blob padded to a page
+ * boundary reads as zeroes past its real size. Every read and skip advances
one position counter,
+ * which drives the clamp and {@link #available()}.
+ */
public class BlobInputStream extends InputStream {
- private InputStream inputStream;
- private long fileSize;
- private long totalRead = 0;
+ private final InputStream inputStream;
+ private final long fileSize;
+ private long position = 0;
+ private long markedPosition = 0;
+ /**
+ * @param fileSize the size known to the caller, used only when the opened
stream reports none; it
+ * can be stale (list cache), while the stream's own properties describe
the version read
+ */
public BlobInputStream(DataLakeFileOpenInputStreamResult inputStream, long
fileSize) {
this.inputStream = inputStream.getInputStream();
- this.fileSize = fileSize;
+ PathProperties properties = inputStream.getProperties();
+ this.fileSize = properties != null ? properties.getFileSize() : fileSize;
+ }
+
+ private long remaining() {
+ return Math.max(0L, fileSize - position);
}
@Override
public int read() throws IOException {
- int c = inputStream.read();
- totalRead++;
- if (totalRead > fileSize) {
+ if (remaining() == 0) {
return -1;
}
+ int c = inputStream.read();
+ if (c >= 0) {
+ position++;
+ }
return c;
}
@Override
public int read(byte[] bytes) throws IOException {
- int readSize = inputStream.read(bytes);
- if (readSize > 0) {
- totalRead += readSize;
- }
- return readSize;
+ return read(bytes, 0, bytes.length);
}
@Override
- public int read(byte[] bytes, int i, int i1) throws IOException {
- int readSize = inputStream.read(bytes, i, i1);
- if (readSize > 0 && totalRead + readSize > fileSize) {
- // This blob is padded to a page boundary
- // Just return the remainder. The rest are 0 anyway
- //
- int actuallyRead = (int) (fileSize - totalRead);
- totalRead = fileSize;
- return actuallyRead;
+ public int read(byte[] bytes, int offset, int length) throws IOException {
+ if (length == 0) {
+ return 0;
+ }
+ long remaining = remaining();
+ if (remaining == 0) {
+ return -1;
+ }
+ int readSize = inputStream.read(bytes, offset, (int) Math.min(length,
remaining));
+ if (readSize > 0) {
+ position += readSize;
}
return readSize;
}
@Override
- public long skip(long l) throws IOException {
- long skippedLength = inputStream.skip(l);
- if (skippedLength > 0) {
- totalRead += skippedLength;
- // This blob is padded to a page boundary
- // Just return the remainder. The rest are 0 anyway
- //
- if (totalRead + skippedLength > fileSize) {
- int actuallyRead = (int) (fileSize - totalRead);
- totalRead = fileSize;
- return actuallyRead;
- }
+ public long skip(long length) throws IOException {
+ if (length <= 0) {
+ return 0;
+ }
+ long skipped = inputStream.skip(Math.min(length, remaining()));
+ if (skipped > 0) {
+ position += skipped;
}
- return skippedLength;
+ return Math.max(0L, skipped);
}
@Override
public int available() throws IOException {
- return (int) (fileSize - totalRead);
+ return (int) Math.min(Integer.MAX_VALUE, remaining());
}
@Override
@@ -94,13 +105,15 @@ public class BlobInputStream extends InputStream {
}
@Override
- public synchronized void mark(int i) {
- inputStream.mark(i);
+ public synchronized void mark(int readLimit) {
+ inputStream.mark(readLimit);
+ markedPosition = position;
}
@Override
public synchronized void reset() throws IOException {
inputStream.reset();
+ position = markedPosition;
}
@Override
diff --git
a/plugins/tech/azure/src/test/java/org/apache/hop/vfs/azure/BlobInputStreamPositionTest.java
b/plugins/tech/azure/src/test/java/org/apache/hop/vfs/azure/BlobInputStreamPositionTest.java
new file mode 100644
index 0000000000..cc2b7079d7
--- /dev/null
+++
b/plugins/tech/azure/src/test/java/org/apache/hop/vfs/azure/BlobInputStreamPositionTest.java
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.vfs.azure;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import
com.azure.storage.file.datalake.models.DataLakeFileOpenInputStreamResult;
+import com.azure.storage.file.datalake.models.PathProperties;
+import java.io.BufferedInputStream;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Position bookkeeping of {@link BlobInputStream}: skip, available and the
clamp to the file size
+ * must agree with the bytes actually consumed from the underlying stream.
+ */
+class BlobInputStreamPositionTest {
+
+ private static final int SIZE = 100;
+
+ private static byte[] content(int length) {
+ byte[] bytes = new byte[length];
+ for (int i = 0; i < length; i++) {
+ bytes[i] = (byte) i;
+ }
+ return bytes;
+ }
+
+ private static BlobInputStream wrap(InputStream source, long fileSize) {
+ DataLakeFileOpenInputStreamResult result =
mock(DataLakeFileOpenInputStreamResult.class);
+ when(result.getInputStream()).thenReturn(source);
+ return new BlobInputStream(result, fileSize);
+ }
+
+ private static BlobInputStream wrap(byte[] bytes) {
+ return wrap(new ByteArrayInputStream(bytes), bytes.length);
+ }
+
+ @Test
+ void bulkReadsReturnTheWholeFile() throws IOException {
+ byte[] bytes = content(SIZE);
+ try (BlobInputStream in = wrap(bytes)) {
+ assertArrayEquals(bytes, in.readAllBytes());
+ }
+ }
+
+ @Test
+ void availableShrinksAfterBulkReads() throws IOException {
+ try (BlobInputStream in = wrap(content(SIZE))) {
+ assertEquals(30, in.read(new byte[30], 0, 30));
+ assertEquals(SIZE - 30, in.available());
+ }
+ }
+
+ @Test
+ void skipReturnsTheNumberOfBytesSkipped() throws IOException {
+ try (BlobInputStream in = wrap(content(SIZE))) {
+ assertEquals(60, in.skip(60));
+ assertEquals(SIZE - 60, in.available());
+ }
+ }
+
+ @Test
+ void bulkReadAfterSkipContinuesAtTheNewPosition() throws IOException {
+ byte[] bytes = content(SIZE);
+ try (BlobInputStream in = wrap(bytes)) {
+ in.skip(60);
+ assertArrayEquals(Arrays.copyOfRange(bytes, 60, SIZE),
in.readAllBytes());
+ }
+ }
+
+ @Test
+ void singleByteReadAfterSkipContinuesAtTheNewPosition() throws IOException {
+ try (BlobInputStream in = wrap(content(SIZE))) {
+ in.skip(60);
+ assertEquals(60, in.read());
+ }
+ }
+
+ @Test
+ void skipPastTheEndStopsAtTheEnd() throws IOException {
+ try (BlobInputStream in = wrap(content(SIZE))) {
+ assertEquals(40, in.read(new byte[40], 0, 40));
+ assertEquals(SIZE - 40, in.skip(1000));
+ assertEquals(0, in.available());
+ assertEquals(-1, in.read());
+ }
+ }
+
+ /** The way VFS consumes it: wrapped in a BufferedInputStream
(FileContentInputStream). */
+ @Test
+ void bufferedSkipThenReadReturnsTheRemainingBytes() throws IOException {
+ byte[] bytes = content(SIZE);
+ try (InputStream in = new BufferedInputStream(wrap(bytes))) {
+ in.skipNBytes(60);
+ assertArrayEquals(Arrays.copyOfRange(bytes, 60, SIZE),
in.readAllBytes());
+ }
+ }
+
+ /** The clamp exists for blobs padded past their real size; padding must
never leak out. */
+ @Test
+ void paddingBeyondTheFileSizeIsNotReturned() throws IOException {
+ byte[] padded = Arrays.copyOf(content(SIZE), 128);
+ try (BlobInputStream in = wrap(new ByteArrayInputStream(padded), SIZE)) {
+ assertArrayEquals(content(SIZE), in.readAllBytes());
+ assertEquals(-1, in.read());
+ }
+ }
+
+ /** A cached size can be older than the file; the opened stream's own size
wins. */
+ @Test
+ void sizeReportedByTheOpenedStreamWinsOverAStaleSize() throws IOException {
+ byte[] bytes = content(SIZE);
+ PathProperties properties = mock(PathProperties.class);
+ when(properties.getFileSize()).thenReturn((long) SIZE);
+ DataLakeFileOpenInputStreamResult result =
mock(DataLakeFileOpenInputStreamResult.class);
+ when(result.getInputStream()).thenReturn(new ByteArrayInputStream(bytes));
+ when(result.getProperties()).thenReturn(properties);
+ try (BlobInputStream in = new BlobInputStream(result, 50)) {
+ assertArrayEquals(bytes, in.readAllBytes());
+ }
+ }
+
+ @Test
+ void paddingIsNotReturnedAcrossSeveralSmallReads() throws IOException {
+ byte[] padded = Arrays.copyOf(content(SIZE), 128);
+ try (BlobInputStream in = wrap(new ByteArrayInputStream(padded), SIZE)) {
+ byte[] buffer = new byte[32];
+ int total = 0;
+ int n;
+ while ((n = in.read(buffer, 0, buffer.length)) > 0) {
+ total += n;
+ }
+ assertEquals(SIZE, total);
+ }
+ }
+}
