address review comments

.palantir/revapi.yml

@@ -1096,12 +1096,6 @@ acceptedBreaks:
     - code: "java.class.removed"
       old: "enum org.apache.iceberg.BaseMetastoreTableOperations.CommitStatus"
       justification: "Removing deprecated code"
-    - code: "java.method.addedToInterface"
-      new: "method java.lang.Long org.apache.iceberg.encryption.NativeEncryptionKeyMetadata::fileLength()"
-      justification: "New method in interface"
-    - code: "java.method.addedToInterface"
-      new: "method org.apache.iceberg.encryption.NativeEncryptionKeyMetadata org.apache.iceberg.encryption.NativeEncryptionKeyMetadata::copyWithLength(long)"
-      justification: "New method in interface"
     - code: "java.method.removed"
       old: "method java.lang.String org.apache.iceberg.FileScanTaskParser::toJson(org.apache.iceberg.FileScanTask)"
       justification: "Removing deprecated code"

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

@@ -160,10 +160,4 @@ private CatalogProperties() {}
 
   public static final String ENCRYPTION_KMS_TYPE = "encryption.kms-type";
   public static final String ENCRYPTION_KMS_IMPL = "encryption.kms-impl";
-  public static final String WRITER_KEK_TIMEOUT_SEC = "encryption.kek-timeout-sec";
-
-  /**
-   * Default time-out of key encryption keys. Per NIST SP 800-57 P1 R5 section 5.3.6, set to 1 week.
-   */
-  public static final long WRITER_KEK_TIMEOUT_SEC_DEFAULT = TimeUnit.DAYS.toSeconds(7);
 }

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

@@ -25,6 +25,7 @@
 import org.apache.iceberg.ManifestListFile;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -77,7 +78,7 @@ public static KeyManagementClient createKmsClient(Map<String, String> catalogPro
 
   /**
    * @deprecated will be removed in 2.0.0. use {@link #createEncryptionManager(String, int,
-   *     KeyManagementClient, long)} instead.
+   *     KeyManagementClient)} instead.
    */
   @Deprecated
   public static EncryptionManager createEncryptionManager(
@@ -89,12 +90,11 @@ public static EncryptionManager createEncryptionManager(
             TableProperties.ENCRYPTION_DEK_LENGTH,
             TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT);
 
-    return createEncryptionManager(
-        tableKeyId, dataKeyLength, kmsClient, CatalogProperties.WRITER_KEK_TIMEOUT_SEC_DEFAULT);
+    return createEncryptionManager(tableKeyId, dataKeyLength, kmsClient);
   }
 
   public static EncryptionManager createEncryptionManager(
-      String tableKeyId, int dataKeyLength, KeyManagementClient kmsClient, long writerKekTimeout) {
+      String tableKeyId, int dataKeyLength, KeyManagementClient kmsClient) {
     Preconditions.checkArgument(kmsClient != null, "Invalid KMS client: null");
 
     if (null == tableKeyId) {
@@ -107,8 +107,7 @@ public static EncryptionManager createEncryptionManager(
         "Invalid data key length: %s (must be 16, 24, or 32)",
         dataKeyLength);
 
-    return new StandardEncryptionManager(
-        tableKeyId, dataKeyLength, ImmutableList.of(), kmsClient, writerKekTimeout);
+    return new StandardEncryptionManager(tableKeyId, dataKeyLength, ImmutableList.of(), kmsClient);
   }
 
   public static EncryptedOutputFile plainAsEncryptedOutput(OutputFile encryptingOutputFile) {

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

@@ -29,13 +29,19 @@ public interface NativeEncryptionKeyMetadata extends EncryptionKeyMetadata {
   ByteBuffer aadPrefix();
 
   /** Encrypted file length */
-  Long fileLength();
+  default Long fileLength() {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " doesn't implement fileLength");
+  }
 
   /**
    * Copy this key metadata and set the file length.
    *
    * @param length length of the encrypted file in bytes
    * @return a copy of this key metadata (key and AAD) with the file length
    */
-  NativeEncryptionKeyMetadata copyWithLength(long length);
+  default NativeEncryptionKeyMetadata copyWithLength(long length) {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " doesn't implement copyWithLength");
+  }
 }

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

@@ -26,7 +26,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
-import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
@@ -38,25 +37,29 @@
 import org.apache.iceberg.util.ByteBuffers;
 
 public class StandardEncryptionManager implements EncryptionManager {
+  /**
+   * Default time-out of key encryption keys. Per NIST SP 800-57 P1 R5 section 5.3.6, set to 1 week.
+   */
+  private static final long KEY_ENCRYPTION_KEY_TIMEOUT = TimeUnit.DAYS.toMillis(7);
+
   private final String tableKeyId;
   private final int dataKeyLength;
-  private final long writerKekTimeout;
 
   // a holder class of metadata that is not available after serialization
   private class TransientEncryptionState {
     private final KeyManagementClient kmsClient;
     private final Map<String, WrappedEncryptionKey> encryptionKeys;
-    private final LoadingCache<ByteBuffer, ByteBuffer> unwrappedKeyCache;
+    private final LoadingCache<String, ByteBuffer> unwrappedKeyCache;
     private WrappedEncryptionKey currentEncryptionKey;
 
     private TransientEncryptionState(KeyManagementClient kmsClient) {
       this.kmsClient = kmsClient;
       this.encryptionKeys = Maps.newLinkedHashMap();
       this.unwrappedKeyCache =
           Caffeine.newBuilder()
-              .maximumSize(1000)
-              .expireAfterWrite(1, TimeUnit.DAYS)
-              .build(wrappedKey -> kmsClient.unwrapKey(wrappedKey, tableKeyId));
+              .expireAfterWrite(1, TimeUnit.HOURS)
+              .build(
+                  keyId -> kmsClient.unwrapKey(encryptionKeys.get(keyId).wrappedKey(), tableKeyId));
       this.currentEncryptionKey = null;
     }
   }
@@ -67,31 +70,24 @@ private TransientEncryptionState(KeyManagementClient kmsClient) {
 
   /**
    * @deprecated will be removed in 1.8.0. use {@link #StandardEncryptionManager(String, int, List,
-   *     KeyManagementClient, long)} instead.
+   *     KeyManagementClient)} instead.
    */
   @Deprecated
   public StandardEncryptionManager(
       String tableKeyId, int dataKeyLength, KeyManagementClient kmsClient) {
-    this(
-        tableKeyId,
-        dataKeyLength,
-        ImmutableList.of(),
-        kmsClient,
-        CatalogProperties.WRITER_KEK_TIMEOUT_SEC_DEFAULT);
+    this(tableKeyId, dataKeyLength, ImmutableList.of(), kmsClient);
   }
 
   /**
    * @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
-   * @param writerKekTimeout timeout of kek (key encryption key) cache entries
    */
   StandardEncryptionManager(
       String tableKeyId,
       int dataKeyLength,
       List<WrappedEncryptionKey> keys,
-      KeyManagementClient kmsClient,
-      long writerKekTimeout) {
+      KeyManagementClient kmsClient) {
     Preconditions.checkNotNull(tableKeyId, "Invalid encryption key ID: null");
     Preconditions.checkArgument(
         dataKeyLength == 16 || dataKeyLength == 24 || dataKeyLength == 32,
@@ -101,7 +97,6 @@ public StandardEncryptionManager(
     this.tableKeyId = tableKeyId;
     this.transientState = new TransientEncryptionState(kmsClient);
     this.dataKeyLength = dataKeyLength;
-    this.writerKekTimeout = writerKekTimeout;
 
     for (WrappedEncryptionKey key : keys) {
       transientState.encryptionKeys.put(key.id(), key);
@@ -159,11 +154,7 @@ public ByteBuffer wrapKey(ByteBuffer secretKey) {
    */
   @Deprecated
   public ByteBuffer unwrapKey(ByteBuffer wrappedSecretKey) {
-    if (transientState == null) {
-      throw new IllegalStateException("Cannot unwrap key after serialization (missing KMS client)");
-    }
-
-    return transientState.unwrappedKeyCache.get(wrappedSecretKey);
+    throw new UnsupportedOperationException("Use unwrapKey(String) instead");
   }
 
   public String currentSnapshotKeyId() {
@@ -173,7 +164,7 @@ public String currentSnapshotKeyId() {
 
     if (transientState.currentEncryptionKey == null
         || transientState.currentEncryptionKey.timestamp()
-            < System.currentTimeMillis() - writerKekTimeout) {
+            < System.currentTimeMillis() - KEY_ENCRYPTION_KEY_TIMEOUT) {
       createNewEncryptionKey();
     }
 
@@ -185,8 +176,7 @@ public ByteBuffer unwrapKey(String keyId) {
       throw new IllegalStateException("Cannot unwrap key after serialization (missing KMS client)");
     }
 
-    return transientState.unwrappedKeyCache.get(
-        transientState.encryptionKeys.get(keyId).wrappedKey());
+    return transientState.unwrappedKeyCache.get(keyId);
   }
 
   private ByteBuffer newKey() {
@@ -212,7 +202,7 @@ private void createNewEncryptionKey() {
         new WrappedEncryptionKey(newKeyId(), wrapped, System.currentTimeMillis());
 
     // update internal tracking
-    transientState.unwrappedKeyCache.put(wrapped, unwrapped);
+    transientState.unwrappedKeyCache.put(key.id(), unwrapped);
     transientState.encryptionKeys.put(key.id(), key);
     transientState.currentEncryptionKey = key;
   }

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

@@ -21,6 +21,10 @@
 import java.io.Serializable;
 import java.nio.ByteBuffer;
 
+/**
+ * This class keeps a wrapped (KMS-encrypted) version of the keys used to encrypt manifest list key
+ * metadata. These keys have an ID and a creation timestamp.
+ */
 public class WrappedEncryptionKey implements Serializable {
   private final String keyID;
   private final ByteBuffer wrappedKey;

core/src/test/java/org/apache/iceberg/encryption/EncryptionTestHelpers.java

@@ -35,7 +35,6 @@ public static EncryptionManager createEncryptionManager() {
     return EncryptionUtil.createEncryptionManager(
         UnitestKMS.MASTER_KEY_NAME1,
         TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT,
-        EncryptionUtil.createKmsClient(catalogProperties),
-        CatalogProperties.WRITER_KEK_TIMEOUT_SEC_DEFAULT);
+        EncryptionUtil.createKmsClient(catalogProperties));
   }
 }