v0 of dynamoDB support

kafka-client/build.gradle.kts

@@ -138,6 +138,7 @@ dependencies {
     implementation(libs.bundles.scylla)
     implementation(libs.bundles.commons)
     implementation(libs.mongodb.driver.sync)
+    implementation(libs.bundles.dynamo)
     implementation(libs.bundles.otel)
     implementation(libs.bundles.grpc)
     implementation(libs.protobuf.java.util)

kafka-client/src/main/java/dev/responsive/kafka/api/ResponsiveKafkaStreams.java

@@ -12,6 +12,7 @@
 
 package dev.responsive.kafka.api;
 
+import static dev.responsive.kafka.api.config.ResponsiveConfig.DYNAMODB_ENDPOINT_CONFIG;
 import static dev.responsive.kafka.api.config.ResponsiveConfig.METRICS_ENABLED_CONFIG;
 import static dev.responsive.kafka.api.config.ResponsiveConfig.MONGO_ADDITIONAL_CONNECTION_STRING_PARAMS_CONFIG;
 import static dev.responsive.kafka.api.config.ResponsiveConfig.MONGO_ENDPOINT_CONFIG;
@@ -41,6 +42,7 @@
 import dev.responsive.kafka.internal.config.ResponsiveStreamsConfig;
 import dev.responsive.kafka.internal.db.CassandraClientFactory;
 import dev.responsive.kafka.internal.db.DefaultCassandraClientFactory;
+import dev.responsive.kafka.internal.db.DynamoClient;
 import dev.responsive.kafka.internal.db.mongo.CollectionCreationOptions;
 import dev.responsive.kafka.internal.db.mongo.ResponsiveMongoClient;
 import dev.responsive.kafka.internal.db.rs3.RS3TableFactory;
@@ -61,6 +63,7 @@
 import dev.responsive.kafka.internal.utils.SessionUtil;
 import java.io.File;
 import java.io.IOException;
+import java.net.URI;
 import java.nio.file.Files;
 import java.time.Duration;
 import java.util.Base64;
@@ -93,6 +96,11 @@
 import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
 
 public class ResponsiveKafkaStreams extends KafkaStreams {
 
@@ -532,23 +540,26 @@ public Params build() {
       final var admin = responsiveKafkaClientSupplier.getAdmin(responsiveConfig.originals());
       if (compatibilityMode == CompatibilityMode.METRICS_ONLY) {
         sessionClients = new SessionClients(
-            Optional.empty(), Optional.empty(), Optional.empty(), false, admin);
+            Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), false, admin);
         return this;
       }
 
       final var backendType = ConfigUtils.storageBackend(responsiveConfig);
       switch (backendType) {
         case CASSANDRA:
+          LOG.info("using Cassandra responsive store");
           final var cqlSession = cassandraFactory.createCqlSession(responsiveConfig, metrics);
           sessionClients = new SessionClients(
               Optional.empty(),
               Optional.of(cassandraFactory.createClient(cqlSession, responsiveConfig)),
               Optional.empty(),
+              Optional.empty(),
               false,
               admin
           );
           break;
         case MONGO_DB:
+          LOG.info("using MongoDB responsive store");
           final var hostname = responsiveConfig.getString(MONGO_ENDPOINT_CONFIG);
           final String clientId = responsiveConfig.getString(MONGO_USERNAME_CONFIG);
           final Password clientSecret = responsiveConfig.getPassword(MONGO_PASSWORD_CONFIG);
@@ -570,6 +581,27 @@ public Params build() {
               )),
               Optional.empty(),
               Optional.empty(),
+              Optional.empty(),
+              false,
+              admin
+          );
+          break;
+        case DYNAMO_DB:
+          LOG.info("using DynamoDB responsive store");
+          final var dynamoEndpoint = responsiveConfig.getString(DYNAMODB_ENDPOINT_CONFIG);
+          final var dynamoDB = DynamoDbAsyncClient.builder()
+              .region(Region.US_WEST_2)
+              // TODO(agavra): support real credentials
+              .credentialsProvider(StaticCredentialsProvider.create(
+                  AwsBasicCredentials.create("dummykey", "dummysecret")))
+              .endpointOverride(URI.create(dynamoEndpoint))
+              .httpClientBuilder(NettyNioAsyncHttpClient.builder())
+              .build();
+          sessionClients = new SessionClients(
+              Optional.empty(),
+              Optional.empty(),
+              Optional.empty(),
+              Optional.of(new DynamoClient(dynamoDB, responsiveConfig)),
               false,
               admin
           );
@@ -580,6 +612,7 @@ public Params build() {
               Optional.empty(),
               Optional.empty(),
               Optional.empty(),
+              Optional.empty(),
               true,
               admin
           );
@@ -592,6 +625,7 @@ public Params build() {
               Optional.empty(),
               Optional.empty(),
               Optional.of(new RS3TableFactory(rs3Host, rs3Port)),
+              Optional.empty(),
               false,
               admin
           );

kafka-client/src/main/java/dev/responsive/kafka/api/config/ResponsiveConfig.java

@@ -119,6 +119,11 @@ public class ResponsiveConfig extends AbstractConfig {
   public static final String RS3_PORT_CONFIG = "responsive.rs3.port";
   private static final String RS3_PORT_DOC = "The port to use when connecting to RS3.";
 
+  // ------------------ DynamoDB specific configurations -----------------------
+
+  public static final String DYNAMODB_ENDPOINT_CONFIG = "responsive.dynamodb.endpoint";
+  private static final String DYNAMODB_ENDPOINT_DOC = "The DynamoDB endpoint to connect to.";
+
   // ------------------ ScyllaDB specific configurations ----------------------
 
   public static final String CASSANDRA_USERNAME_CONFIG = "responsive.cassandra.username";
@@ -370,6 +375,16 @@ public class ResponsiveConfig extends AbstractConfig {
           MONGO_ENDPOINT_DOC
       )
 
+      // dyanmodb connection configurations
+      .define(
+          DYNAMODB_ENDPOINT_CONFIG,
+          Type.STRING,
+          null,
+          new ConfigDef.NonEmptyString(),
+          Importance.HIGH,
+          DYNAMODB_ENDPOINT_DOC
+      )
+
       // cassandra connection configurations
       .define(
           CASSANDRA_USERNAME_CONFIG,

kafka-client/src/main/java/dev/responsive/kafka/api/config/StorageBackend.java

@@ -17,6 +17,7 @@
 public enum StorageBackend {
   CASSANDRA,
   MONGO_DB,
+  DYNAMO_DB,
   IN_MEMORY,
   RS3;
 

kafka-client/src/main/java/dev/responsive/kafka/internal/db/DynamoClient.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2024 Responsive Computing, Inc.
+ *
+ * This source code is licensed under the Responsive Business Source License Agreement v1.0
+ * available at:
+ *
+ * https://www.responsive.dev/legal/responsive-bsl-10
+ *
+ * This software requires a valid Commercial License Key for production use. Trial and commercial
+ * licenses can be obtained at https://www.responsive.dev
+ */
+
+package dev.responsive.kafka.internal.db;
+
+import dev.responsive.kafka.api.config.ResponsiveConfig;
+import dev.responsive.kafka.internal.db.dynamo.DynamoKVTable;
+import dev.responsive.kafka.internal.db.partitioning.TablePartitioner;
+import dev.responsive.kafka.internal.db.spec.DefaultTableSpec;
+import dev.responsive.kafka.internal.stores.TtlResolver;
+import java.util.Optional;
+import java.util.concurrent.TimeoutException;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
+
+public class DynamoClient {
+
+  private final ResponsiveConfig responsiveConfig;
+  private final TableCache<DynamoKVTable> kvTableCache;
+
+
+  public DynamoClient(
+      final DynamoDbAsyncClient dynamoDB,
+      final ResponsiveConfig responsiveConfig
+  ) {
+    this.responsiveConfig = responsiveConfig;
+    kvTableCache = new TableCache<>(
+        spec -> new DynamoKVTable(
+            dynamoDB,
+            spec.tableName()
+        ));
+  }
+
+  public DynamoKVTable kvTable(
+      final String name,
+      final Optional<TtlResolver<?, ?>> ttlResolver
+  ) throws InterruptedException, TimeoutException {
+    return kvTableCache.create(
+        new DefaultTableSpec(name, TablePartitioner.defaultPartitioner(), ttlResolver)
+    );
+  }
+}

kafka-client/src/main/java/dev/responsive/kafka/internal/db/dynamo/DynamoKVFlushManager.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2024 Responsive Computing, Inc.
+ *
+ * This source code is licensed under the Responsive Business Source License Agreement v1.0
+ * available at:
+ *
+ * https://www.responsive.dev/legal/responsive-bsl-10
+ *
+ * This software requires a valid Commercial License Key for production use. Trial and commercial
+ * licenses can be obtained at https://www.responsive.dev
+ */
+
+package dev.responsive.kafka.internal.db.dynamo;
+
+import dev.responsive.kafka.internal.db.KVFlushManager;
+import dev.responsive.kafka.internal.db.RemoteWriter;
+import dev.responsive.kafka.internal.db.partitioning.TablePartitioner;
+import dev.responsive.kafka.internal.stores.RemoteWriteResult;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
+
+public class DynamoKVFlushManager extends KVFlushManager {
+
+  private final String logPrefix;
+  private final Logger log;
+
+  private final DynamoKVTable table;
+
+  private final int kafkaPartition;
+  private final DynamoDbAsyncClient dynamo;
+  private final TablePartitioner<Bytes, Integer> partitioner;
+
+
+  public DynamoKVFlushManager(
+      final DynamoKVTable table,
+      final int kafkaPartition,
+      final DynamoDbAsyncClient dynamo
+  ) {
+    this.table = table;
+    this.kafkaPartition = kafkaPartition;
+    this.dynamo = dynamo;
+
+    partitioner = TablePartitioner.defaultPartitioner();
+    logPrefix = String.format("%s[%d] kv-store {epoch=%d} ",
+        table.name(), kafkaPartition, table.localEpoch(kafkaPartition));
+    log = new LogContext(logPrefix).logger(DynamoKVFlushManager.class);
+  }
+
+  @Override
+  public String tableName() {
+    return table.name();
+  }
+
+  @Override
+  public TablePartitioner<Bytes, Integer> partitioner() {
+    return partitioner;
+  }
+
+  @Override
+  public RemoteWriter<Bytes, Integer> createWriter(
+      final Integer tablePartition,
+      final long consumedOffset
+  ) {
+    return new DynamoWriter<>(table, kafkaPartition, tablePartition, dynamo);
+  }
+
+  @Override
+  public String failedFlushInfo(final long batchOffset, final Integer failedTablePartition) {
+    // TODO(agavra): add persisted epoch
+    return String.format("<batchOffset=%d, persistedOffset=%d>, <localEpoch=%d, persistedEpoch=%d>",
+        batchOffset, table.fetchOffset(kafkaPartition),
+        table.localEpoch(kafkaPartition), 0);
+  }
+
+  @Override
+  public String logPrefix() {
+    return logPrefix;
+  }
+
+  @Override
+  public RemoteWriteResult<Integer> updateOffset(final long consumedOffset) {
+    try {
+      table.setOffset(kafkaPartition, consumedOffset);
+      return RemoteWriteResult.success(kafkaPartition);
+    } catch (final Exception e) {
+      // TODO(agavra): improve error handling to be closer to MongoKVFlushManager
+      log.error("Failed to update offset", e);
+      return RemoteWriteResult.failure(kafkaPartition);
+    }
+  }
+}