Aliyun/Dell: Classify missing object reads as NotFoundException

aliyun/src/main/java/org/apache/iceberg/aliyun/oss/OSSInputStream.java

@@ -19,10 +19,13 @@
 package org.apache.iceberg.aliyun.oss;
 
 import com.aliyun.oss.OSS;
+import com.aliyun.oss.OSSErrorCode;
+import com.aliyun.oss.OSSException;
 import com.aliyun.oss.model.GetObjectRequest;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Arrays;
+import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.FileIOMetricsContext;
 import org.apache.iceberg.io.SeekableInputStream;
 import org.apache.iceberg.metrics.Counter;
@@ -147,7 +150,16 @@ private void openStream() throws IOException {
     closeStream();
 
     GetObjectRequest request = new GetObjectRequest(uri.bucket(), uri.key()).withRange(pos, -1);
-    stream = client.getObject(request).getObjectContent();
+    try {
+      stream = client.getObject(request).getObjectContent();
+    } catch (OSSException e) {
+      if (OSSErrorCode.NO_SUCH_BUCKET.equals(e.getErrorCode())
+          || OSSErrorCode.NO_SUCH_KEY.equals(e.getErrorCode())) {
+        throw new NotFoundException(e, "Location does not exist: %s", uri);
+      }
+
+      throw e;
+    }
   }
 
   private void closeStream() throws IOException {

aliyun/src/test/java/org/apache/iceberg/aliyun/oss/TestOSSInputStream.java

@@ -26,6 +26,7 @@
 import java.util.Arrays;
 import java.util.Random;
 import java.util.concurrent.ThreadLocalRandom;
+import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.SeekableInputStream;
 import org.apache.iceberg.relocated.com.google.common.io.ByteStreams;
 import org.junit.jupiter.api.Test;
@@ -119,6 +120,17 @@ public void testSeek() throws Exception {
     }
   }
 
+  @Test
+  void missingObjectThrowsNotFoundException() throws Exception {
+    OSSURI uri = new OSSURI(location("missing.dat"));
+
+    try (SeekableInputStream in = new OSSInputStream(ossClient().get(), uri)) {
+      assertThatThrownBy(in::read)
+          .isInstanceOf(NotFoundException.class)
+          .hasMessageContaining(uri.location());
+    }
+  }
+
   private byte[] randomData(int size) {
     byte[] data = new byte[size];
     random.nextBytes(data);

dell/src/main/java/org/apache/iceberg/dell/ecs/EcsSeekableInputStream.java

@@ -20,8 +20,10 @@
 
 import com.emc.object.Range;
 import com.emc.object.s3.S3Client;
+import com.emc.object.s3.S3Exception;
 import java.io.IOException;
 import java.io.InputStream;
+import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.FileIOMetricsContext;
 import org.apache.iceberg.io.SeekableInputStream;
 import org.apache.iceberg.metrics.Counter;
@@ -110,7 +112,15 @@ private void checkAndUseNewPos() throws IOException {
     }
 
     pos = newPos;
-    internalStream = client.readObjectStream(uri.bucket(), uri.name(), Range.fromOffset(pos));
+    try {
+      internalStream = client.readObjectStream(uri.bucket(), uri.name(), Range.fromOffset(pos));
+    } catch (S3Exception e) {
+      if (e.getHttpCode() == 404) {
+        throw new NotFoundException(e, "Location does not exist: %s", uri);
+      }
+
+      throw e;
+    }
     newPos = -1;
   }
 

dell/src/test/java/org/apache/iceberg/dell/ecs/TestEcsSeekableInputStream.java

@@ -19,11 +19,13 @@
 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;
 import java.nio.charset.StandardCharsets;
 import org.apache.iceberg.dell.mock.ecs.EcsS3MockRule;
+import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.metrics.MetricsContext;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -90,4 +92,17 @@ public void testReadBytes() throws IOException {
           .isEqualTo("012");
     }
   }
+
+  @Test
+  void missingObjectThrowsNotFoundException() throws IOException {
+    String objectName = rule.randomObjectName();
+    EcsURI uri = new EcsURI(rule.bucket(), objectName);
+
+    try (EcsSeekableInputStream input =
+        new EcsSeekableInputStream(rule.client(), uri, MetricsContext.nullMetrics())) {
+      assertThatThrownBy(input::read)
+          .isInstanceOf(NotFoundException.class)
+          .hasMessageContaining(uri.location());
+    }
+  }
 }