Separate StorageCredential/Credential

Author: nastra

aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java

@@ -105,7 +105,7 @@ public class S3FileIO
   private MetricsContext metrics = MetricsContext.nullMetrics();
   private final AtomicBoolean isResourceClosed = new AtomicBoolean(false);
   private transient StackTraceElement[] createStack;
-  private List<? extends StorageCredential> storageCredentials = ImmutableList.of();
+  private List<StorageCredential> storageCredentials = ImmutableList.of();
 
   /**
    * No-arg constructor to load the FileIO dynamically.
@@ -565,13 +565,13 @@ private boolean recoverObject(ObjectVersion version, String bucket) {
   }
 
   @Override
-  public void setCredentials(List<? extends StorageCredential> credentials) {
+  public void setCredentials(List<StorageCredential> credentials) {
     Preconditions.checkArgument(credentials != null, "Invalid storage credentials: null");
     this.storageCredentials = credentials;
   }
 
   @Override
-  public List<? extends StorageCredential> credentials() {
+  public List<StorageCredential> credentials() {
     return ImmutableList.copyOf(storageCredentials);
   }
 

core/src/main/java/org/apache/iceberg/CatalogUtil.java

@@ -372,7 +372,7 @@ public static FileIO loadFileIO(
       String impl,
       Map<String, String> properties,
       Object hadoopConf,
-      List<? extends StorageCredential> storageCredentials) {
+      List<StorageCredential> storageCredentials) {
     LOG.info("Loading custom FileIO implementation: {}", impl);
     DynConstructors.Ctor<FileIO> ctor;
     try {

core/src/main/java/org/apache/iceberg/io/SupportsStorageCredentials.java

@@ -22,7 +22,7 @@
 
 public interface SupportsStorageCredentials {
 
-  void setCredentials(List<? extends StorageCredential> credentials);
+  void setCredentials(List<StorageCredential> credentials);
 
-  List<? extends StorageCredential> credentials();
+  List<StorageCredential> credentials();
 }

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -27,6 +27,7 @@
 import java.util.Set;
 import java.util.function.BiFunction;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
@@ -54,6 +55,8 @@
 import org.apache.iceberg.io.CloseableGroup;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.FileIOTracker;
+import org.apache.iceberg.io.ImmutableStorageCredential;
+import org.apache.iceberg.io.StorageCredential;
 import org.apache.iceberg.metrics.MetricsReporter;
 import org.apache.iceberg.metrics.MetricsReporters;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -972,7 +975,16 @@ private FileIO newFileIO(
       return ioBuilder.apply(context, properties);
     } else {
       String ioImpl = properties.getOrDefault(CatalogProperties.FILE_IO_IMPL, DEFAULT_FILE_IO_IMPL);
-      return CatalogUtil.loadFileIO(ioImpl, properties, conf, storageCredentials);
+      List<StorageCredential> credentials =
+          storageCredentials.stream()
+              .map(
+                  c ->
+                      ImmutableStorageCredential.builder()
+                          .prefix(c.prefix())
+                          .config(c.config())
+                          .build())
+              .collect(Collectors.toList());
+      return CatalogUtil.loadFileIO(ioImpl, properties, conf, credentials);
     }
   }
 

core/src/main/java/org/apache/iceberg/rest/credentials/Credential.java

@@ -18,9 +18,19 @@
  */
 package org.apache.iceberg.rest.credentials;
 
-import org.apache.iceberg.io.StorageCredential;
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.immutables.value.Value;
 
-@SuppressWarnings("immutables:incompat")
 @Value.Immutable
-public interface Credential extends StorageCredential {}
+public interface Credential {
+  String prefix();
+
+  Map<String, String> config();
+
+  @Value.Check
+  default void validate() {
+    Preconditions.checkArgument(!prefix().isEmpty(), "Invalid prefix: must be non-empty");
+    Preconditions.checkArgument(!config().isEmpty(), "Invalid config: must be non-empty");
+  }
+}

core/src/test/java/org/apache/iceberg/TestCatalogUtil.java

@@ -522,7 +522,7 @@ public void report(MetricsReport report) {}
   public static class TestFileIOWithStorageCredentials
       implements FileIO, SupportsStorageCredentials {
 
-    private List<? extends StorageCredential> storageCredentials;
+    private List<StorageCredential> storageCredentials;
 
     public TestFileIOWithStorageCredentials() {}
 
@@ -540,12 +540,12 @@ public OutputFile newOutputFile(String path) {
     public void deleteFile(String path) {}
 
     @Override
-    public void setCredentials(List<? extends StorageCredential> credentials) {
+    public void setCredentials(List<StorageCredential> credentials) {
       this.storageCredentials = credentials;
     }
 
     @Override
-    public List<? extends StorageCredential> credentials() {
+    public List<StorageCredential> credentials() {
       return storageCredentials;
     }
   }

gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java

@@ -75,7 +75,7 @@ public class GCSFileIO implements DelegateFileIO, SupportsStorageCredentials {
   private final AtomicBoolean isResourceClosed = new AtomicBoolean(false);
   private SerializableMap<String, String> properties = null;
   private OAuth2RefreshCredentialsHandler refreshHandler = null;
-  private List<? extends StorageCredential> storageCredentials = ImmutableList.of();
+  private List<StorageCredential> storageCredentials = ImmutableList.of();
 
   /**
    * No-arg constructor to load the FileIO dynamically.
@@ -263,13 +263,13 @@ private void internalDeleteFiles(Stream<BlobId> blobIdsToDelete) {
   }
 
   @Override
-  public void setCredentials(List<? extends StorageCredential> credentials) {
+  public void setCredentials(List<StorageCredential> credentials) {
     Preconditions.checkArgument(credentials != null, "Invalid storage credentials: null");
     this.storageCredentials = credentials;
   }
 
   @Override
-  public List<? extends StorageCredential> credentials() {
+  public List<StorageCredential> credentials() {
     return ImmutableList.copyOf(storageCredentials);
   }