Add Encryption When Writing to Iceberg Tables in RecordWriter.java (#34021)

Author: SaumilPatel03

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriter.java

@@ -29,6 +29,8 @@
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
 import org.apache.iceberg.io.DataWriter;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.OutputFile;
@@ -60,7 +62,6 @@ class RecordWriter {
       throws IOException {
     this.table = table;
     this.fileFormat = fileFormat;
-
     MetricsConfig metricsConfig = MetricsConfig.forTable(table);
 
     if (table.spec().isUnpartitioned()) {
@@ -72,8 +73,12 @@ class RecordWriter {
               table.locationProvider().newDataLocation(table.spec(), partitionKey, filename));
     }
     OutputFile outputFile;
+    EncryptionKeyMetadata keyMetadata;
     try (FileIO io = table.io()) {
-      outputFile = io.newOutputFile(absoluteFilename);
+      OutputFile tmpFile = io.newOutputFile(absoluteFilename);
+      EncryptedOutputFile encryptedOutputFile = table.encryption().encrypt(tmpFile);
+      outputFile = encryptedOutputFile.encryptingOutputFile();
+      keyMetadata = encryptedOutputFile.keyMetadata();
     }
 
     switch (fileFormat) {
@@ -85,6 +90,7 @@ class RecordWriter {
                 .withSpec(table.spec())
                 .withPartition(partitionKey)
                 .metricsConfig(metricsConfig)
+                .withKeyMetadata(keyMetadata)
                 .overwrite()
                 .build();
         break;
@@ -96,6 +102,7 @@ class RecordWriter {
                 .withSpec(table.spec())
                 .withPartition(partitionKey)
                 .metricsConfig(metricsConfig)
+                .withKeyMetadata(keyMetadata)
                 .overwrite()
                 .build();
         break;
@@ -114,6 +121,7 @@ class RecordWriter {
   }
 
   public void write(Record record) {
+
     icebergDataWriter.write(record);
   }