SSEC integration changes

rajdchak · rajdchak · commit 73da94ada686 · 2025-06-12T13:59:36.000+01:00
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -1920,7 +1920,8 @@ private FSDataInputStream executeOpen(
         .withCallbacks(createInputStreamCallbacks(auditSpan))
         .withContext(readContext.build())
         .withObjectAttributes(createObjectAttributes(path, fileStatus))
-        .withStreamStatistics(inputStreamStats);
+        .withStreamStatistics(inputStreamStats)
+        .withEncryptionSecrets(getEncryptionSecrets());
     return new FSDataInputStream(getStore().readObject(parameters));
   }
 
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/AnalyticsStream.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/AnalyticsStream.java
@@ -21,9 +21,13 @@
 
 import java.io.EOFException;
 import java.io.IOException;
+import java.util.Optional;
 
+import org.apache.hadoop.fs.s3a.S3AEncryptionMethods;
+import org.apache.hadoop.fs.s3a.auth.delegation.EncryptionSecretOperations;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamFactory;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStream;
+import software.amazon.s3.analyticsaccelerator.request.EncryptionSecrets;
 import software.amazon.s3.analyticsaccelerator.request.ObjectMetadata;
 import software.amazon.s3.analyticsaccelerator.util.InputPolicy;
 import software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation;
@@ -205,6 +209,12 @@ private OpenStreamInformation buildOpenStreamInformation(ObjectReadParameters pa
           .etag(parameters.getObjectAttributes().getETag()).build());
     }
 
+    if(parameters.getEncryptionSecrets().getEncryptionMethod() == S3AEncryptionMethods.SSE_C) {
+      EncryptionSecretOperations.getSSECustomerKey(parameters.getEncryptionSecrets())
+              .ifPresent(base64customerKey -> openStreamInformationBuilder.encryptionSecrets(
+              EncryptionSecrets.builder().sseCustomerKey(Optional.of(base64customerKey)).build()));
+    }
+
     return openStreamInformationBuilder.build();
   }
 
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/ObjectReadParameters.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/ObjectReadParameters.java
@@ -23,6 +23,7 @@
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.s3a.S3AReadOpContext;
 import org.apache.hadoop.fs.s3a.S3ObjectAttributes;
+import org.apache.hadoop.fs.s3a.auth.delegation.EncryptionSecrets;
 import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
 
 import static java.util.Objects.requireNonNull;
@@ -69,6 +70,29 @@ public final class ObjectReadParameters {
    */
   private LocalDirAllocator directoryAllocator;
 
+  /**
+   * Encryption secrets for this stream
+   */
+  private EncryptionSecrets encryptionSecrets;
+
+  /**
+   * Getter.
+   * @return Encryption secrets.
+   */
+  public EncryptionSecrets getEncryptionSecrets() {
+    return encryptionSecrets;
+  }
+
+  /**
+   * Set encryption secrets.
+   * @param value new value
+   * @return the builder
+   */
+  public ObjectReadParameters withEncryptionSecrets(final EncryptionSecrets value) {
+    encryptionSecrets = value;
+    return this;
+  }
+
   /**
    * @return Read operation context.
    */
@@ -185,6 +209,7 @@ public ObjectReadParameters validate() {
     requireNonNull(directoryAllocator, "directoryAllocator");
     requireNonNull(objectAttributes, "objectAttributes");
     requireNonNull(streamStatistics, "streamStatistics");
+    requireNonNull(encryptionSecrets, "encryptionSecrets");
     return this;
   }
 }
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AAnalyticsAcceleratorStreamReading.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AAnalyticsAcceleratorStreamReading.java
@@ -22,7 +22,10 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.lang.reflect.Field;
 
+import org.apache.hadoop.fs.contract.s3a.S3AContract;
+import org.apache.hadoop.fs.s3a.impl.streams.AnalyticsStream;
 import org.junit.Before;
 import org.junit.Test;
 import org.assertj.core.api.Assertions;
@@ -36,14 +39,21 @@
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStream;
 import org.apache.hadoop.fs.statistics.IOStatistics;
 
+import software.amazon.s3.analyticsaccelerator.S3SeekableInputStream;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamConfiguration;
 import software.amazon.s3.analyticsaccelerator.common.ConnectorConfiguration;
+import software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation;
 
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_PARQUET;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.writeDataset;
 import static org.apache.hadoop.fs.s3a.Constants.ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX;
+import static org.apache.hadoop.fs.s3a.Constants.S3_ENCRYPTION_ALGORITHM;
+import static org.apache.hadoop.fs.s3a.Constants.S3_ENCRYPTION_KEY;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.enableAnalyticsAccelerator;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestBucketName;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
 import static org.apache.hadoop.fs.s3a.test.PublicDatasetTestUtils.getExternalData;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
@@ -64,6 +74,7 @@
 public class ITestS3AAnalyticsAcceleratorStreamReading extends AbstractS3ATestBase {
 
   private static final String PHYSICAL_IO_PREFIX = "physicalio";
+  private static final String SSEC_KEY = "4niV/jPK5VFRHY+KNb6wtqYd4xXyMgdJ9XQJpcQUVbs=";
 
   private Path externalTestFile;
 
@@ -194,4 +205,61 @@ public void testInvalidConfigurationThrows() throws Exception {
         () -> S3SeekableInputStreamConfiguration.fromConfiguration(connectorConfiguration));
   }
 
+  /**
+   * This test verifies that the OpenStreamInfo object contains correct encryption
+   * settings when reading an SSEC encrypted file with analytics stream.
+   *
+   * @throws Exception
+   */
+
+  @Test
+  public void testAnalyticsStreamOpenStreamInfoWhenSSECEncryptionEnabled() throws Exception {
+    Configuration confSSEC = this.createConfiguration();
+    S3ATestUtils.disableFilesystemCaching(confSSEC);
+    removeBaseAndBucketOverrides(getTestBucketName(confSSEC), confSSEC, S3_ENCRYPTION_ALGORITHM, S3_ENCRYPTION_KEY);
+
+    confSSEC.set(S3_ENCRYPTION_ALGORITHM, S3AEncryptionMethods.SSE_C.getMethod());
+    confSSEC.set(S3_ENCRYPTION_KEY, SSEC_KEY);
+
+    S3AContract contractSSEC = (S3AContract) createContract(confSSEC);
+    contractSSEC.init();
+    contractSSEC.setConf(confSSEC);
+    S3AFileSystem fileSystemSSEC = (S3AFileSystem) contractSSEC.getTestFileSystem();
+
+    int len = TEST_FILE_LEN;
+    describe("Create an encrypted file and verify OpenStreamInfo encryption settings");
+    Path src = methodPath();
+    byte[] data = dataset(len, 'a', 'z');
+    writeDataset(fileSystemSSEC, src, data, len, 1024 * 1024, true);
+    // Read with encryption settings
+    OpenStreamInformation originalInfo;
+    try (FSDataInputStream in = fileSystemSSEC.open(src)) {
+      AnalyticsStream s3AInputStream = (AnalyticsStream) in.getWrappedStream();
+      Field inputStreamField = s3AInputStream.getClass().getDeclaredField("inputStream");
+      inputStreamField.setAccessible(true);
+      S3SeekableInputStream seekableInputStream = (S3SeekableInputStream) inputStreamField.get(s3AInputStream);
+
+      Field logicalIOField = seekableInputStream.getClass().getDeclaredField("logicalIO");
+      logicalIOField.setAccessible(true);
+      Object logicalIO = logicalIOField.get(seekableInputStream);
+
+      Field physicalIOField = logicalIO.getClass().getDeclaredField("physicalIO");
+      physicalIOField.setAccessible(true);
+      Object physicalIO = physicalIOField.get(logicalIO);
+
+      Field openStreamInfoField = physicalIO.getClass().getDeclaredField("openStreamInformation");
+      openStreamInfoField.setAccessible(true);
+      originalInfo = (OpenStreamInformation) openStreamInfoField.get(physicalIO);
+
+      // Verify the OpenStreamInfo object exists
+      assertNotNull(originalInfo.getEncryptionSecrets().getSsecCustomerKey().get(), "OpenStreamInfo should not be null");
+      assertEquals(SSEC_KEY, originalInfo.getEncryptionSecrets().getSsecCustomerKey().get());
+    }
+
+    // Clean up
+    fileSystemSSEC.delete(src, false);
+  }
+
+
+
 }
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEncryptionSSEC.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEncryptionSSEC.java
@@ -43,7 +43,6 @@
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestBucketName;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.maybeSkipRootTests;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfAnalyticsAcceleratorEnabled;
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
@@ -96,8 +95,6 @@ protected Configuration createConfiguration() {
   @Override
   public void setup() throws Exception {
     super.setup();
-    skipIfAnalyticsAcceleratorEnabled(getConfiguration(),
-        "Analytics Accelerator currently does not support SSE-C");
     assumeEnabled();
     // although not a root dir test, this confuses paths enough it shouldn't be run in
     // parallel with other jobs
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesSSECDiskBlocks.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesSSECDiskBlocks.java
@@ -31,7 +31,6 @@
 import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_ALGORITHM;
 import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_KEY;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfAnalyticsAcceleratorEnabled;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfEncryptionTestsDisabled;
 
 /**
@@ -55,8 +54,6 @@ public class ITestS3AHugeFilesSSECDiskBlocks
   public void setup() throws Exception {
     try {
       super.setup();
-      skipIfAnalyticsAcceleratorEnabled(getConfiguration(),
-          "Analytics Accelerator currently does not support SSE-C");
     } catch (AccessDeniedException | AWSUnsupportedFeatureException e) {
       skip("Bucket does not allow " + S3AEncryptionMethods.SSE_C + " encryption method");
     }

Original file line number	Diff line number	Diff line change
`@@ -1920,7 +1920,8 @@ private FSDataInputStream executeOpen(`
`1920`	`1920`	`.withCallbacks(createInputStreamCallbacks(auditSpan))`
`1921`	`1921`	`.withContext(readContext.build())`
`1922`	`1922`	`.withObjectAttributes(createObjectAttributes(path, fileStatus))`
`1923`		`- .withStreamStatistics(inputStreamStats);`
	`1923`	`+ .withStreamStatistics(inputStreamStats)`
	`1924`	`+ .withEncryptionSecrets(getEncryptionSecrets());`
`1924`	`1925`	`return new FSDataInputStream(getStore().readObject(parameters));`
`1925`	`1926`	`}`
`1926`	`1927`