Add conditional write logging to HadoopS3AccessHelper

- Add comprehensive logging to track Hadoop 3.4.2 conditional writes
- Log PutObjectOptions creation and usage in all S3 operations
- Log multipart upload initiation, completion, and single object uploads
- Helps verify conditional write functionality is being invoked
- All logs prefixed with '=== CONDITIONAL WRITES:' for easy filtering

Author: Paul Ashley

flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/HadoopS3AccessHelper.java

@@ -177,7 +177,13 @@ public software.amazon.awssdk.services.s3.model.UploadPartResponse uploadPart(
 
     /** Creates default PutObjectOptions for Hadoop 3.4.2. */
     private static org.apache.hadoop.fs.s3a.impl.PutObjectOptions createDefaultPutObjectOptions() {
-        return org.apache.hadoop.fs.s3a.impl.PutObjectOptions.keepingDirs();
+        org.apache.hadoop.fs.s3a.impl.PutObjectOptions options = org.apache.hadoop.fs.s3a.impl.PutObjectOptions.keepingDirs();
+        
+        // Log conditional write configuration
+        LOG.info("=== CONDITIONAL WRITES: Creating PutObjectOptions with keepingDirs() - Hadoop 3.4.2 ===");
+        LOG.info("=== PutObjectOptions details: {} ===", options.toString());
+        
+        return options;
     }
 
     /**
@@ -413,11 +419,18 @@ private static void validateOutputFile(File file) {
     public String startMultiPartUpload(String key) throws IOException {
         validateKey(key);
         try {
+            LOG.info("=== CONDITIONAL WRITES: Initiating multipart upload for key: {} ===", key);
+            
             // Hadoop 3.4.2 uses AWS SDK v2 and requires PutObjectOptions
-            String uploadId =
-                    s3accessHelper.initiateMultiPartUpload(key, createDefaultPutObjectOptions());
+            org.apache.hadoop.fs.s3a.impl.PutObjectOptions putObjectOptions = createDefaultPutObjectOptions();
+            LOG.info("=== CONDITIONAL WRITES: Using PutObjectOptions: {} ===", putObjectOptions);
+            
+            String uploadId = s3accessHelper.initiateMultiPartUpload(key, putObjectOptions);
+            
+            LOG.info("=== CONDITIONAL WRITES: Multipart upload initiated successfully. UploadId: {} ===", uploadId);
             return uploadId;
         } catch (Exception e) {
+            LOG.error("=== CONDITIONAL WRITES: Failed to initiate multipart upload for key: {} - Error: {} ===", key, e.getMessage());
             throw e;
         }
     }
@@ -481,6 +494,10 @@ public UploadPartResult uploadPart(
     public PutObjectResult putObject(String key, File inputFile) throws IOException {
         validateKey(key);
         validateInputFile(inputFile);
+        
+        LOG.info("=== CONDITIONAL WRITES: Putting object for key: {}, file size: {} bytes ===", 
+                key, inputFile.length());
+        
         // Hadoop 3.4.2 uses AWS SDK v2 with different put object API
         // Create AWS SDK v2 PutObjectRequest with correct bucket name
         software.amazon.awssdk.services.s3.model.PutObjectRequest putRequest =
@@ -493,6 +510,8 @@ public PutObjectResult putObject(String key, File inputFile) throws IOException
         // Create PutObjectOptions
         org.apache.hadoop.fs.s3a.impl.PutObjectOptions putObjectOptions =
                 createDefaultPutObjectOptions();
+        
+        LOG.info("=== CONDITIONAL WRITES: Using PutObjectOptions for putObject: {} ===", putObjectOptions);
 
         // Note: For Hadoop 3.4.2, the putObject API with BlockUploadData is designed for
         // block-based uploads from memory. For file-based uploads, it's more appropriate
@@ -532,6 +551,9 @@ public PutObjectResult putObject(String key, File inputFile) throws IOException
                 result.setSSECustomerAlgorithm(response.sseCustomerAlgorithm());
             }
 
+            LOG.info("=== CONDITIONAL WRITES: PutObject completed successfully for key: {}, ETag: {} ===", 
+                    key, result.getETag());
+            
             return result;
 
         } catch (software.amazon.awssdk.core.exception.SdkException e) {
@@ -572,6 +594,12 @@ public CompleteMultipartUploadResult commitMultiPartUpload(
                                                 .build())
                         .collect(java.util.stream.Collectors.toList());
 
+        LOG.info("=== CONDITIONAL WRITES: Completing multipart upload for key: {}, uploadId: {}, parts: {} ===", 
+                destKey, uploadId, partETags.size());
+        
+        org.apache.hadoop.fs.s3a.impl.PutObjectOptions putObjectOptions = createDefaultPutObjectOptions();
+        LOG.info("=== CONDITIONAL WRITES: Using PutObjectOptions for completion: {} ===", putObjectOptions);
+        
         // Use the new completeMPUwithRetries API
         software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse response =
                 s3accessHelper.completeMPUwithRetries(
@@ -580,7 +608,10 @@ public CompleteMultipartUploadResult commitMultiPartUpload(
                         completedParts,
                         length,
                         errorCount,
-                        createDefaultPutObjectOptions());
+                        putObjectOptions);
+        
+        LOG.info("=== CONDITIONAL WRITES: Multipart upload completed successfully for key: {}, ETag: {} ===", 
+                destKey, response.eTag());
 
         // Convert AWS SDK v2 response to AWS SDK v1 response
         CompleteMultipartUploadResult result = new CompleteMultipartUploadResult();

flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3FileSystemFactory.java

@@ -37,6 +37,23 @@
 public class S3FileSystemFactory extends AbstractS3FileSystemFactory {
 
     private static final Logger LOG = LoggerFactory.getLogger(S3FileSystemFactory.class);
+    
+    static {
+        // Static initializer - this will log when the class is first loaded
+        String timestamp = java.time.Instant.now().toString();
+        String message = String.format("=== CUSTOM S3FileSystemFactory LOADED [%s] - Hadoop 3.4.2 + Conditional Writes ===", timestamp);
+        System.err.println(message);
+        LOG.error(message);
+        
+        // Also log the JAR location to help with debugging
+        try {
+            String jarLocation = S3FileSystemFactory.class.getProtectionDomain().getCodeSource().getLocation().toString();
+            LOG.error("=== JAR Location: {} ===", jarLocation);
+            System.err.println("=== JAR Location: " + jarLocation + " ===");
+        } catch (Exception e) {
+            LOG.error("=== Could not determine JAR location: {} ===", e.getMessage());
+        }
+    }
 
     private static final String[] FLINK_CONFIG_PREFIXES = {"s3.", "s3a.", "fs.s3a."};
 
@@ -45,10 +62,15 @@ public class S3FileSystemFactory extends AbstractS3FileSystemFactory {
         {"fs.s3a.secret-key", "fs.s3a.secret.key"},
         {"fs.s3a.path-style-access", "fs.s3a.path.style.access"},
         {"fs.s3a.requester-pays-enabled", "fs.s3a.requester.pays.enabled"},
+        {"fs.s3a.create-conditional-enabled", "fs.s3a.create.conditional.enabled"},
+        {"s3a.create-conditional-enabled", "fs.s3a.create.conditional.enabled"}
     };
 
     public S3FileSystemFactory() {
         super("Hadoop s3a file system", createHadoopConfigLoader());
+        
+        // Distinctive log message to confirm our custom JAR is loaded
+        LOG.error("=== CUSTOM S3 FILESYSTEM JAR LOADED - Hadoop 3.4.2 + Conditional Writes ===");
     }
 
     @Override