diff --git a/airbyte-config/config-persistence/build.gradle b/airbyte-config/config-persistence/build.gradle
index 824882d95d60..7d1738018db0 100644
--- a/airbyte-config/config-persistence/build.gradle
+++ b/airbyte-config/config-persistence/build.gradle
@@ -20,6 +20,7 @@ dependencies {
implementation 'commons-io:commons-io:2.7'
implementation 'com.google.cloud:google-cloud-secretmanager:2.0.5'
implementation 'com.bettercloud:vault-java-driver:5.1.0'
+ implementation 'com.amazonaws.secretsmanager:aws-secretsmanager-caching-java:1.0.1'
testImplementation 'org.hamcrest:hamcrest-all:1.3'
testImplementation libs.platform.testcontainers.postgresql
diff --git a/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/AWSSecretManagerPersistence.java b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/AWSSecretManagerPersistence.java
new file mode 100644
index 000000000000..417abee4d305
--- /dev/null
+++ b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/AWSSecretManagerPersistence.java
@@ -0,0 +1,105 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config.persistence.split_secrets;
+
+import com.amazonaws.secretsmanager.caching.SecretCache;
+import com.amazonaws.services.secretsmanager.AWSSecretsManager;
+import com.amazonaws.services.secretsmanager.AWSSecretsManagerClientBuilder;
+import com.amazonaws.services.secretsmanager.model.CreateSecretRequest;
+import com.amazonaws.services.secretsmanager.model.DeleteSecretRequest;
+import com.amazonaws.services.secretsmanager.model.ResourceNotFoundException;
+import com.amazonaws.services.secretsmanager.model.UpdateSecretRequest;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * SecretPersistence implementation for AWS Secret Manager using Java
+ * SDK The current implementation doesn't make use of `SecretCoordinate#getVersion` as this
+ * version is non-compatible with how AWS secret manager deals with versions. In AWS versions is an
+ * internal idiom that can is accessible, but it's a UUID + a tag more
+ * details.
+ */
+@Slf4j
+public class AWSSecretManagerPersistence implements SecretPersistence {
+
+ private final AWSSecretsManager client;
+
+ @VisibleForTesting
+ protected final SecretCache cache;
+
+ /**
+ * Creates a AWSSecretManagerPersistence using the defaults client and region from the current AWS
+ * credentials. This implementation makes use of SecretCache as optimization to access secrets
+ *
+ * @see SecretCache
+ */
+ public AWSSecretManagerPersistence() {
+ this.client = AWSSecretsManagerClientBuilder.defaultClient();
+ this.cache = new SecretCache(this.client);
+ }
+
+ /**
+ * Creates a AWSSecretManagerPersistence overriding the current region. This implementation makes
+ * use of SecretCache as optimization to access secrets
+ *
+ * @param region AWS region to use
+ * @see SecretCache
+ */
+ public AWSSecretManagerPersistence(final String region) {
+ this.client = AWSSecretsManagerClientBuilder
+ .standard()
+ .withRegion(region)
+ .build();
+ this.cache = new SecretCache(this.client);
+ }
+
+ @Override
+ public Optional read(final SecretCoordinate coordinate) {
+ String secretString = null;
+ try {
+ log.debug("Reading secret {}", coordinate.getCoordinateBase());
+ secretString = cache.getSecretString(coordinate.getCoordinateBase());
+ } catch (ResourceNotFoundException e) {
+ log.warn("Secret {} not found", coordinate.getCoordinateBase());
+ }
+ return Optional.ofNullable(secretString);
+ }
+
+ @Override
+ public void write(final SecretCoordinate coordinate, final String payload) {
+ if (read(coordinate).isPresent()) {
+ log.debug("Secret {} found updating payload.", coordinate.getCoordinateBase());
+ final UpdateSecretRequest request = new UpdateSecretRequest()
+ .withSecretId(coordinate.getCoordinateBase())
+ .withSecretString(payload)
+ .withDescription("Airbyte secret.");
+ client.updateSecret(request);
+ } else {
+ log.debug("Secret {} not found, creating a new one.", coordinate.getCoordinateBase());
+ final CreateSecretRequest secretRequest = new CreateSecretRequest()
+ .withName(coordinate.getCoordinateBase())
+ .withSecretString(payload)
+ .withDescription("Airbyte secret.");
+ client.createSecret(secretRequest);
+ }
+
+ }
+
+ /**
+ * Utility to clean up after integration tests.
+ *
+ * @param coordinate SecretCoordinate to delete.
+ */
+ @VisibleForTesting
+ protected void deleteSecret(final SecretCoordinate coordinate) {
+ client.deleteSecret(new DeleteSecretRequest()
+ .withSecretId(coordinate.getCoordinateBase())
+ .withForceDeleteWithoutRecovery(true));
+ }
+
+}
diff --git a/airbyte-config/config-persistence/src/test-integration/java/io/airbyte/config/persistence/split_secrets/AWSSecretManagerPersistenceIntegrationTest.java b/airbyte-config/config-persistence/src/test-integration/java/io/airbyte/config/persistence/split_secrets/AWSSecretManagerPersistenceIntegrationTest.java
new file mode 100644
index 000000000000..3c0bc568dedf
--- /dev/null
+++ b/airbyte-config/config-persistence/src/test-integration/java/io/airbyte/config/persistence/split_secrets/AWSSecretManagerPersistenceIntegrationTest.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config.persistence.split_secrets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Optional;
+import org.apache.commons.lang3.RandomUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class AWSSecretManagerPersistenceIntegrationTest {
+
+ public String coordinate_base;
+ private AWSSecretManagerPersistence persistence;
+
+ @BeforeEach
+ void setup() {
+ persistence = new AWSSecretManagerPersistence();
+ coordinate_base = "aws/airbyte/secret/integration/" + RandomUtils.nextInt() % 20000;
+ }
+
+ @Test
+ void testReadWriteUpdate() throws InterruptedException {
+ SecretCoordinate secretCoordinate = new SecretCoordinate(coordinate_base, 1);
+
+ // try reading a non-existent secret
+ Optional firstRead = persistence.read(secretCoordinate);
+ assertTrue(firstRead.isEmpty());
+
+ // write it
+ String payload = "foo-secret";
+ persistence.write(secretCoordinate, payload);
+ persistence.cache.refreshNow(secretCoordinate.getCoordinateBase());
+ Optional read2 = persistence.read(secretCoordinate);
+ assertTrue(read2.isPresent());
+ assertEquals(payload, read2.get());
+
+ // update it
+ final var secondPayload = "bar-secret";
+ final var coordinate2 = new SecretCoordinate(coordinate_base, 2);
+ persistence.write(coordinate2, secondPayload);
+ persistence.cache.refreshNow(secretCoordinate.getCoordinateBase());
+ final var thirdRead = persistence.read(coordinate2);
+ assertTrue(thirdRead.isPresent());
+ assertEquals(secondPayload, thirdRead.get());
+ }
+
+ @AfterEach
+ void tearDown() {
+ persistence.deleteSecret(new SecretCoordinate(coordinate_base, 1));
+ }
+
+}