s3 retry option is added

airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/storage/StorageClient.kt

@@ -21,6 +21,8 @@ import io.micronaut.kotlin.context.createBean
 import jakarta.inject.Inject
 import jakarta.inject.Singleton
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials
+import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration
+import software.amazon.awssdk.core.retry.RetryPolicy
 import software.amazon.awssdk.core.sync.RequestBody
 import software.amazon.awssdk.regions.Region
 import software.amazon.awssdk.services.s3.S3Client
@@ -314,7 +316,15 @@ internal fun MinioStorageConfig.s3Client(): S3Client =
  */
 internal fun S3StorageConfig.s3Client(): S3Client {
   val builder = S3Client.builder().region(Region.of(this.region))
-
+  // Adding more try to avoid unknown host exception
+  val retryPolicy = RetryPolicy.builder()
+  if (!this.retry.isNullOrBlank()) {
+    retryPolicy.numRetries(this.retry!!.toInt())
+  } else {
+    retryPolicy.numRetries(25)
+  }
+  val clientOverrideConfiguration = ClientOverrideConfiguration.builder().retryPolicy(retryPolicy.build())
+  builder.overrideConfiguration(clientOverrideConfiguration.build())
   // If credentials are part of this config, specify them. Otherwise, let the SDK default credential provider take over.
   if (!this.accessKey.isNullOrBlank()) {
     builder.credentialsProvider {

airbyte-commons-worker/src/test/kotlin/io/airbyte/workers/storage/StorageClientTest.kt

@@ -252,7 +252,7 @@ class MinioStorageClientTest {
 }
 
 class S3StorageClientTest {
-  private val config = S3StorageConfig(buckets = buckets, accessKey = "access", secretAccessKey = "secret", region = "us-east-1")
+  private val config = S3StorageConfig(buckets = buckets, accessKey = "access", secretAccessKey = "secret", region = "us-east-1", retry = "10")
 
   @Test
   fun `key matches`() {

airbyte-commons/src/main/kotlin/io/airbyte/commons/envvar/EnvVar.kt

@@ -14,6 +14,7 @@ enum class EnvVar {
   AWS_ASSUME_ROLE_SECRET_ACCESS_KEY,
   AWS_ASSUME_ROLE_SECRET_NAME,
   AWS_DEFAULT_REGION,
+  AWS_S3_RETRY,
   AWS_SECRET_ACCESS_KEY,
 
   CDK_ENTRYPOINT,

airbyte-config/config-models/src/main/java/io/airbyte/config/EnvConfigs.java

@@ -105,7 +105,8 @@ private StorageConfig getLogConfiguration() {
           buckets,
           getEnsureEnv(EnvVar.AWS_ACCESS_KEY_ID),
           getEnsureEnv(EnvVar.AWS_ACCESS_KEY_ID),
-          getEnsureEnv(EnvVar.AWS_DEFAULT_REGION));
+          getEnsureEnv(EnvVar.AWS_DEFAULT_REGION),
+          getEnsureEnv(EnvVar.AWS_S3_RETRY));
       default -> throw new IllegalArgumentException(EnvVar.STORAGE_TYPE.name() + " has is an unsupported value");
     };
   }

airbyte-config/config-models/src/main/kotlin/io/airbyte/config/storage/StorageConfig.kt

@@ -86,6 +86,7 @@ data class S3StorageConfig(
   @Value("\${$STORAGE_S3.access-key}") val accessKey: String?,
   @Value("\${$STORAGE_S3.secret-access-key}") val secretAccessKey: String?,
   @Value("\${$STORAGE_S3.region}") val region: String,
+  @Value("\${$STORAGE_S3.retry}") val retry: String?,
 ) : StorageConfig {
   override fun toEnvVarMap(): Map<String, String> =
     buildMap {
@@ -100,6 +101,9 @@ data class S3StorageConfig(
       secretAccessKey?.let {
         put(EnvVar.AWS_SECRET_ACCESS_KEY, secretAccessKey)
       }
+      retry?.let {
+        put(EnvVar.AWS_S3_RETRY, retry)
+      }
       put(EnvVar.AWS_DEFAULT_REGION, region)
     }.mapKeys { it.key.name }
 

airbyte-config/config-models/src/test/java/io/airbyte/config/helpers/S3LogsTest.java

@@ -43,7 +43,7 @@ class S3LogsTest {
         new StorageBucketConfig(bucketLog, "state", "workload", "payload"),
         EnvVar.AWS_ACCESS_KEY_ID.fetch(),
         EnvVar.AWS_SECRET_ACCESS_KEY.fetch(),
-        region));
+        region, "10"));
   }
 
   private S3Client s3Client;

airbyte-config/config-models/src/test/java/io/airbyte/config/storage/DefaultS3ClientFactoryTest.java

@@ -13,7 +13,7 @@ class DefaultS3ClientFactoryTest {
   @Test
   void testS3() {
     final var bucket = new StorageBucketConfig("log", "state", "workload", "payload");
-    final var config = new S3StorageConfig(bucket, "access-key", "access-key-secret", "us-east-1");
+    final var config = new S3StorageConfig(bucket, "access-key", "access-key-secret", "us-east-1", "10");
 
     assertDoesNotThrow(() -> new DefaultS3ClientFactory(config).get());
   }

airbyte-config/config-models/src/test/kotlin/io/airbyte/config/helpers/CloudLogsTest.kt

@@ -20,7 +20,7 @@ class CloudLogsTest {
 
   @Test
   fun `createCloudLogClient for s3`() {
-    val storageConfig = S3StorageConfig(buckets = buckets, accessKey = "ak", secretAccessKey = "sak", region = "r")
+    val storageConfig = S3StorageConfig(buckets = buckets, accessKey = "ak", secretAccessKey = "sak", region = "r", retry = "10")
     val cfg = LogConfigs(storageConfig)
 
     assertTrue(CloudLogs.createCloudLogClient(cfg) is S3Logs)

airbyte-config/config-models/src/test/kotlin/io/airbyte/config/storage/StorageConfigTest.kt

@@ -183,6 +183,7 @@ class StorageConfigTest {
         accessKey = private,
         secretAccessKey = private,
         region = public,
+        retry = "10",
       )
 
     with(gcs.toString()) {

deps.toml

@@ -59,8 +59,8 @@ apache-cxf-core = { module = "org.apache.cxf:cxf-core", version = "3.4.2" }
 apache-mime4j-core = { module = "org.apache.james:apache-mime4j-core", version = "0.8.10" } # Transitive dependency of keycloak. Forced upgrade to 0.8.10 from 0.8.9
 appender-log4j2 = { module = "com.therealvan:appender-log4j2", version = "5.3.2" }
 assertj-core = { module = "org.assertj:assertj-core", version = "3.21.0" }
-aws-java-sdk-s3 = { module = "com.amazonaws:aws-java-sdk-s3", version = "1.12.701" }
-aws-java-sdk-sts = {module = "com.amazonaws:aws-java-sdk-sts", version = "1.12.701"}
+aws-java-sdk-s3 = { module = "com.amazonaws:aws-java-sdk-s3", version = "1.12.749" }
+aws-java-sdk-sts = {module = "com.amazonaws:aws-java-sdk-sts", version = "1.12.749"}
 aws-secretsmanager-caching-java = { module = "com.amazonaws.secretsmanager:aws-secretsmanager-caching-java", version = "1.0.2" }
 bouncycastle-bcprov = { module = "org.bouncycastle:bcprov-jdk15on", version.ref = "bouncycastle" }
 bouncycastle-bcpkix = { module = "org.bouncycastle:bcpkix-jdk15on", version.ref = "bouncycastle" }
@@ -171,8 +171,8 @@ quartz-scheduler = { module = "org.quartz-scheduler:quartz", version = "2.3.2" }
 reactor-core = { module = "io.projectreactor:reactor-core", version.ref = "reactor" }
 reactor-kotlin-extensions = { module = "io.projectreactor.kotlin:reactor-kotlin-extensions", version = "1.2.2" }
 reactor-test = { module = "io.projectreactor:reactor-test", version.ref = "reactor" }
-s3 = { module = "software.amazon.awssdk:s3", version = "2.23.17" }
-sts = { module = "software.amazon.awssdk:sts", version = "2.23.17" }
+s3 = { module = "software.amazon.awssdk:s3", version = "2.26.8" }
+sts = { module = "software.amazon.awssdk:sts", version = "2.26.8" }
 segment-java-analytics = { module = "com.segment.analytics.java:analytics", version.ref = "segment" }
 sendgrid-java = { module = "com.sendgrid:sendgrid-java", version = "4.0.1" }
 sentry-java = { module = "io.sentry:sentry", version.ref = "sentry" }