(iceberg) branch main updated: Dell: Validate ECS stream seeks and closed state (#17962)

[singhpk234]

This is an automated email from the ASF dual-hosted git repository.

singhpk234 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new e4bcaa40cb Dell: Validate ECS stream seeks and closed state (#17962)
e4bcaa40cb is described below

commit e4bcaa40cbb4de6431584cb2a5816cb626c34ca5
Author: Charlie Chen <jchen...@uw.edu>
AuthorDate: Sun Sep 27 17:34:32 2026 -0700

    Dell: Validate ECS stream seeks and closed state (#17962)
    
    Generated-by: Codex
---
 .../iceberg/dell/ecs/EcsSeekableInputStream.java   |  6 +++
 .../dell/ecs/TestEcsSeekableInputStream.java       | 51 ++++++++++++++++++++++
 2 files changed, 57 insertions(+)

diff --git 
a/dell/src/main/java/org/apache/iceberg/dell/ecs/EcsSeekableInputStream.java 
b/dell/src/main/java/org/apache/iceberg/dell/ecs/EcsSeekableInputStream.java
index 6d32d3ba98..ea8b4e7dd7 100644
--- a/dell/src/main/java/org/apache/iceberg/dell/ecs/EcsSeekableInputStream.java
+++ b/dell/src/main/java/org/apache/iceberg/dell/ecs/EcsSeekableInputStream.java
@@ -27,6 +27,7 @@ import org.apache.iceberg.io.SeekableInputStream;
 import org.apache.iceberg.metrics.Counter;
 import org.apache.iceberg.metrics.MetricsContext;
 import org.apache.iceberg.metrics.MetricsContext.Unit;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
 /**
  * A {@link SeekableInputStream} implementation that warp {@link 
S3Client#readObjectStream(String,
@@ -51,6 +52,7 @@ class EcsSeekableInputStream extends SeekableInputStream {
   private long pos = -1;
 
   private InputStream internalStream;
+  private boolean closed;
 
   private final Counter readBytes;
   private final Counter readOperations;
@@ -69,6 +71,8 @@ class EcsSeekableInputStream extends SeekableInputStream {
 
   @Override
   public void seek(long inputNewPos) {
+    Preconditions.checkState(!closed, "already closed");
+    Preconditions.checkArgument(inputNewPos >= 0, "position is negative: %s", 
inputNewPos);
     if (pos == inputNewPos) {
       return;
     }
@@ -96,6 +100,7 @@ class EcsSeekableInputStream extends SeekableInputStream {
   }
 
   private void checkAndUseNewPos() throws IOException {
+    Preconditions.checkState(!closed, "Cannot read: already closed");
     if (newPos < 0) {
       return;
     }
@@ -116,6 +121,7 @@ class EcsSeekableInputStream extends SeekableInputStream {
 
   @Override
   public void close() throws IOException {
+    closed = true;
     if (internalStream != null) {
       internalStream.close();
     }
diff --git 
a/dell/src/test/java/org/apache/iceberg/dell/ecs/TestEcsSeekableInputStream.java
 
b/dell/src/test/java/org/apache/iceberg/dell/ecs/TestEcsSeekableInputStream.java
index de97aa8d6f..154a4a0765 100644
--- 
a/dell/src/test/java/org/apache/iceberg/dell/ecs/TestEcsSeekableInputStream.java
+++ 
b/dell/src/test/java/org/apache/iceberg/dell/ecs/TestEcsSeekableInputStream.java
@@ -19,6 +19,7 @@
 package org.apache.iceberg.dell.ecs;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.emc.object.s3.request.PutObjectRequest;
 import java.io.IOException;
@@ -27,6 +28,8 @@ import org.apache.iceberg.dell.mock.ecs.EcsS3MockRule;
 import org.apache.iceberg.metrics.MetricsContext;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 public class TestEcsSeekableInputStream {
 
@@ -90,4 +93,52 @@ public class TestEcsSeekableInputStream {
           .isEqualTo("012");
     }
   }
+
+  @Test
+  void seekRejectsNegativePositionWithoutChangingState() throws IOException {
+    String objectName = rule.randomObjectName();
+    rule.client()
+        .putObject(new PutObjectRequest(rule.bucket(), objectName, 
"0123456789".getBytes()));
+
+    try (EcsSeekableInputStream input =
+        new EcsSeekableInputStream(
+            rule.client(), new EcsURI(rule.bucket(), objectName), 
MetricsContext.nullMetrics())) {
+      input.seek(2);
+      assertThatThrownBy(() -> input.seek(-1))
+          .isInstanceOf(IllegalArgumentException.class)
+          .hasMessage("position is negative: -1");
+      assertThat(input.getPos()).isEqualTo(2);
+      assertThat(input.read()).isEqualTo('2');
+    }
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  void rejectsSeekAndReadAfterClose(boolean readBeforeClose) throws 
IOException {
+    String objectName = rule.randomObjectName();
+    rule.client()
+        .putObject(new PutObjectRequest(rule.bucket(), objectName, 
"0123456789".getBytes()));
+
+    try (EcsSeekableInputStream input =
+        new EcsSeekableInputStream(
+            rule.client(), new EcsURI(rule.bucket(), objectName), 
MetricsContext.nullMetrics())) {
+      if (readBeforeClose) {
+        assertThat(input.read()).isEqualTo('0');
+      }
+
+      long position = input.getPos();
+      input.close();
+
+      assertThatThrownBy(() -> input.seek(2))
+          .isInstanceOf(IllegalStateException.class)
+          .hasMessage("already closed");
+      assertThatThrownBy(input::read)
+          .isInstanceOf(IllegalStateException.class)
+          .hasMessage("Cannot read: already closed");
+      assertThatThrownBy(() -> input.read(new byte[1], 0, 1))
+          .isInstanceOf(IllegalStateException.class)
+          .hasMessage("Cannot read: already closed");
+      assertThat(input.getPos()).isEqualTo(position);
+    }
+  }
 }