Revert "Encryption integration and test (apache#13066)"

This reverts commit 30ca573.

Author: geruh

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

@@ -214,7 +214,7 @@ protected void refreshFromMetadataLocation(
     this.shouldRefresh = false;
   }
 
-  protected String metadataFileLocation(TableMetadata metadata, String filename) {
+  private String metadataFileLocation(TableMetadata metadata, String filename) {
     String metadataLocation = metadata.properties().get(TableProperties.WRITE_METADATA_LOCATION);
 
     if (metadataLocation != null) {

core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java

@@ -75,7 +75,7 @@ public static KeyManagementClient createKmsClient(Map<String, String> catalogPro
     return kmsClient;
   }
 
-  public static EncryptionManager createEncryptionManager(
+  static EncryptionManager createEncryptionManager(
       List<EncryptedKey> keys, Map<String, String> tableProperties, KeyManagementClient kmsClient) {
     Preconditions.checkArgument(kmsClient != null, "Invalid KMS client: null");
     String tableKeyId = tableProperties.get(TableProperties.ENCRYPTION_TABLE_KEY);
@@ -96,7 +96,7 @@ public static EncryptionManager createEncryptionManager(
         "Invalid data key length: %s (must be 16, 24, or 32)",
         dataKeyLength);
 
-    return new StandardEncryptionManager(keys, tableKeyId, dataKeyLength, kmsClient);
+    return new StandardEncryptionManager(tableKeyId, dataKeyLength, kmsClient);
   }
 
   public static EncryptedOutputFile plainAsEncryptedOutput(OutputFile encryptingOutputFile) {
@@ -128,14 +128,6 @@ public static ByteBuffer decryptManifestListKeyMetadata(
     return ByteBuffer.wrap(decryptedKeyMetadata);
   }
 
-  public static Map<String, EncryptedKey> encryptionKeys(EncryptionManager em) {
-    Preconditions.checkState(
-        em instanceof StandardEncryptionManager,
-        "Retrieving encryption keys requires a StandardEncryptionManager");
-    StandardEncryptionManager sem = (StandardEncryptionManager) em;
-    return sem.encryptionKeys();
-  }
-
   /**
    * Encrypts the key metadata for a manifest list.
    *

core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java

@@ -23,7 +23,6 @@
 import java.nio.ByteBuffer;
 import java.security.SecureRandom;
 import java.util.Base64;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import org.apache.iceberg.TableProperties;
@@ -47,19 +46,9 @@ private class TransientEncryptionState {
     private final Map<String, EncryptedKey> encryptionKeys;
     private final LoadingCache<String, ByteBuffer> unwrappedKeyCache;
 
-    private TransientEncryptionState(KeyManagementClient kmsClient, List<EncryptedKey> keys) {
+    private TransientEncryptionState(KeyManagementClient kmsClient) {
       this.kmsClient = kmsClient;
       this.encryptionKeys = Maps.newLinkedHashMap();
-
-      if (keys != null) {
-        for (EncryptedKey key : keys) {
-          encryptionKeys.put(
-              key.keyId(),
-              new BaseEncryptedKey(
-                  key.keyId(), key.encryptedKeyMetadata(), key.encryptedById(), key.properties()));
-        }
-      }
-
       this.unwrappedKeyCache =
           Caffeine.newBuilder()
               .expireAfterWrite(1, TimeUnit.HOURS)
@@ -75,33 +64,20 @@ private TransientEncryptionState(KeyManagementClient kmsClient, List<EncryptedKe
   private transient volatile SecureRandom lazyRNG = null;
 
   /**
-   * @deprecated will be removed in 2.0.
-   */
-  @Deprecated
-  public StandardEncryptionManager(
-      String tableKeyId, int dataKeyLength, KeyManagementClient kmsClient) {
-    this(List.of(), tableKeyId, dataKeyLength, kmsClient);
-  }
-
-  /**
-   * @param keys encryption keys from table metadata
    * @param tableKeyId table encryption key id
    * @param dataKeyLength length of data encryption key (16/24/32 bytes)
    * @param kmsClient Client of KMS used to wrap/unwrap keys in envelope encryption
    */
   public StandardEncryptionManager(
-      List<EncryptedKey> keys,
-      String tableKeyId,
-      int dataKeyLength,
-      KeyManagementClient kmsClient) {
+      String tableKeyId, int dataKeyLength, KeyManagementClient kmsClient) {
     Preconditions.checkNotNull(tableKeyId, "Invalid encryption key ID: null");
     Preconditions.checkArgument(
         dataKeyLength == 16 || dataKeyLength == 24 || dataKeyLength == 32,
         "Invalid data key length: %s (must be 16, 24, or 32)",
         dataKeyLength);
     Preconditions.checkNotNull(kmsClient, "Invalid KMS client: null");
     this.tableKeyId = tableKeyId;
-    this.transientState = new TransientEncryptionState(kmsClient, keys);
+    this.transientState = new TransientEncryptionState(kmsClient);
     this.dataKeyLength = dataKeyLength;
   }
 
@@ -158,14 +134,6 @@ public ByteBuffer unwrapKey(ByteBuffer wrappedSecretKey) {
     return transientState.kmsClient.unwrapKey(wrappedSecretKey, tableKeyId);
   }
 
-  Map<String, EncryptedKey> encryptionKeys() {
-    if (transientState == null) {
-      throw new IllegalStateException("Cannot return the encryption keys after serialization");
-    }
-
-    return transientState.encryptionKeys;
-  }
-
   private String keyEncryptionKeyID() {
     if (transientState == null) {
       throw new IllegalStateException("Cannot return the current key after serialization");

hive-metastore/src/main/java/org/apache/iceberg/hive/HiveCatalog.java

@@ -18,7 +18,6 @@
  */
 package org.apache.iceberg.hive;
 
-import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -46,8 +45,6 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.encryption.EncryptionUtil;
-import org.apache.iceberg.encryption.KeyManagementClient;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchIcebergViewException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
@@ -91,7 +88,6 @@ public class HiveCatalog extends BaseMetastoreViewCatalog
   private String name;
   private Configuration conf;
   private FileIO fileIO;
-  private KeyManagementClient keyManagementClient;
   private ClientPool<IMetaStoreClient, TException> clients;
   private boolean listAllTables = false;
   private Map<String, String> catalogProperties;
@@ -126,10 +122,6 @@ public void initialize(String inputName, Map<String, String> properties) {
             ? new HadoopFileIO(conf)
             : CatalogUtil.loadFileIO(fileIOImpl, properties, conf);
 
-    if (catalogProperties.containsKey(CatalogProperties.ENCRYPTION_KMS_IMPL)) {
-      this.keyManagementClient = EncryptionUtil.createKmsClient(properties);
-    }
-
     this.clients = new CachedClientPool(conf, properties);
   }
 
@@ -694,8 +686,7 @@ private boolean isValidateNamespace(Namespace namespace) {
   public TableOperations newTableOps(TableIdentifier tableIdentifier) {
     String dbName = tableIdentifier.namespace().level(0);
     String tableName = tableIdentifier.name();
-    return new HiveTableOperations(
-        conf, clients, fileIO, keyManagementClient, name, dbName, tableName);
+    return new HiveTableOperations(conf, clients, fileIO, name, dbName, tableName);
   }
 
   @Override
@@ -825,15 +816,6 @@ protected Map<String, String> properties() {
     return catalogProperties == null ? ImmutableMap.of() : catalogProperties;
   }
 
-  @Override
-  public void close() throws IOException {
-    super.close();
-
-    if (keyManagementClient != null) {
-      keyManagementClient.close();
-    }
-  }
-
   @VisibleForTesting
   void setListAllTables(boolean listAllTables) {
     this.listAllTables = listAllTables;