Fix S3 configuration fields to use Optional<String>

- Change s3_bucket, s3_prefix, s3_region, s3_access_key_id, s3_secret_access_key from String to Optional<String>
- Remove @ConfigDefault("null") annotations as they are not needed for Optional types
- Update usage in transaction method to handle Optional values properly
- Update README.md to clarify that S3 credentials are optional and use default AWS credentials provider chain

Author: kentoyoshida

README.md

@@ -34,8 +34,8 @@ Snowflake output plugin for Embulk loads records to Snowflake.
 - **s3_bucket**: S3 bucket name for JDBC log upload (string, required when upload_jdbc_log_to_s3 is true)
 - **s3_prefix**: S3 key prefix for JDBC log upload (string, required when upload_jdbc_log_to_s3 is true)
 - **s3_region**: AWS region for S3 bucket (string, required when upload_jdbc_log_to_s3 is true)
-- **s3_access_key_id**: AWS access key ID for S3 access (string, optional - uses IAM role if not specified)
-- **s3_secret_access_key**: AWS secret access key for S3 access (string, optional - uses IAM role if not specified)
+- **s3_access_key_id**: AWS access key ID for S3 access (string, optional - uses default AWS credentials provider chain if not specified)
+- **s3_secret_access_key**: AWS secret access key for S3 access (string, optional - uses default AWS credentials provider chain if not specified)
 - **default_timezone**: If input column type (embulk type) is timestamp, this plugin needs to format the timestamp into a SQL string. This default_timezone option is used to control the timezone. You can overwrite timezone for each columns using column_options option. (string, default: `UTC`)
 - **column_options**: advanced: a key-value pairs where key is a column name and value is options for the column.
   - **type**: type of a column when this plugin creates new tables (e.g. `VARCHAR(255)`, `INTEGER NOT NULL UNIQUE`). This used when this plugin creates intermediate tables (insert, truncate_insert and merge modes), when it creates the target table (insert_direct and replace modes), and when it creates nonexistent target table automatically. (string, default: depends on input column type. `BIGINT` if input column type is long, `BOOLEAN` if boolean, `DOUBLE PRECISION` if double, `CLOB` if string, `TIMESTAMP` if timestamp)

src/main/java/org/embulk/output/SnowflakeOutputPlugin.java

@@ -98,24 +98,19 @@ public interface SnowflakePluginTask extends PluginTask {
     public boolean getUploadJdbcLogToS3();
 
     @Config("s3_bucket")
-    @ConfigDefault("null")
-    public String getS3Bucket();
+    public Optional<String> getS3Bucket();
 
     @Config("s3_prefix")
-    @ConfigDefault("null")
-    public String getS3Prefix();
+    public Optional<String> getS3Prefix();
 
     @Config("s3_region")
-    @ConfigDefault("null")
-    public String getS3Region();
+    public Optional<String> getS3Region();
 
     @Config("s3_access_key_id")
-    @ConfigDefault("null")
-    public String getS3AccessKeyId();
+    public Optional<String> getS3AccessKeyId();
 
     @Config("s3_secret_access_key")
-    @ConfigDefault("null")
-    public String getS3SecretAccessKey();
+    public Optional<String> getS3SecretAccessKey();
 
     public void setCopyIntoTableColumnNames(String[] columnNames);
 
@@ -260,15 +255,15 @@ public ConfigDiff transaction(
         if (message != null
             && message.contains(ENCOUNTERED_COMMUNICATION_ERROR_MESSAGE)
             && t.getUploadJdbcLogToS3() == true) {
-          final String s3Bucket = t.getS3Bucket();
-          final String s3Prefix = t.getS3Prefix();
-          final String s3Region = t.getS3Region();
-          final String s3AccessKeyId = t.getS3AccessKeyId();
-          final String s3SecretAccessKey = t.getS3SecretAccessKey();
-          if (s3Bucket == null || s3Prefix == null || s3Region == null) {
+          final Optional<String> s3Bucket = t.getS3Bucket();
+          final Optional<String> s3Prefix = t.getS3Prefix();
+          final Optional<String> s3Region = t.getS3Region();
+          final Optional<String> s3AccessKeyId = t.getS3AccessKeyId();
+          final Optional<String> s3SecretAccessKey = t.getS3SecretAccessKey();
+          if (!s3Bucket.isPresent() || !s3Prefix.isPresent() || !s3Region.isPresent()) {
             logger.warn("s3_bucket, s3_prefix, and s3_region must be set when upload_jdbc_log_to_s3 is true");
           } else {
-            try (JdbcLogUploader jdbcLogUploader = new JdbcLogUploader(s3Bucket, s3Prefix, s3Region, s3AccessKeyId, s3SecretAccessKey)) {
+            try (JdbcLogUploader jdbcLogUploader = new JdbcLogUploader(s3Bucket.get(), s3Prefix.get(), s3Region.get(), s3AccessKeyId.orElse(null), s3SecretAccessKey.orElse(null))) {
               // snowflake_jdbc*.log で最新のファイルを探してアップロード
               String tmpDir = System.getProperty("java.io.tmpdir", "/tmp");
               File logDir = new File(tmpDir);