merge with upstream

responsivedev · Oct 28, 2024 · cc48d46 · cc48d46
2 parents 4a2cea0 + 853518b
commit cc48d46
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 4 deletions.
diff --git a/kafka-client/src/main/java/dev/responsive/kafka/internal/db/CassandraKeyValueTable.java b/kafka-client/src/main/java/dev/responsive/kafka/internal/db/CassandraKeyValueTable.java
@@ -274,7 +274,7 @@ private static CreateTableWithOptions createTable(
         .withColumn(DATA_VALUE.column(), DataTypes.BLOB)
         .withColumn(OFFSET.column(), DataTypes.BIGINT)
         .withColumn(EPOCH.column(), DataTypes.BIGINT)
-        .withColumn(TIMESTAMP.column(), DataTypes.BIGINT);
+        .withColumn(TIMESTAMP.column(), DataTypes.TIMESTAMP);
 
     if (ttlResolver.isPresent() && ttlResolver.get().defaultTtl().isFinite()) {
       final int defaultTtlSeconds = (int) ttlResolver.get().defaultTtl().toSeconds();

diff --git a/...ent/src/test/java/dev/responsive/kafka/internal/db/CassandraFactTableIntegrationTest.java b/...ent/src/test/java/dev/responsive/kafka/internal/db/CassandraFactTableIntegrationTest.java
@@ -129,7 +129,7 @@ public void shouldInsertAndDelete() throws Exception {
 
   @SuppressWarnings("OptionalGetWithoutIsPresent")
   @Test
-  public void shouldRespectSemanticDefaultOnlyTtl() throws Exception {
+  public void shouldConfigureDefaultTtl() throws Exception {
     // Given:
     final var defaultTtl = Duration.ofMinutes(30);
     final var ttlProvider = TtlProvider.<String, String>withDefault(defaultTtl);

diff --git a/...lient/src/test/java/dev/responsive/kafka/internal/db/CassandraKVTableIntegrationTest.java b/...lient/src/test/java/dev/responsive/kafka/internal/db/CassandraKVTableIntegrationTest.java
@@ -20,6 +20,7 @@
 import static dev.responsive.kafka.testutils.IntegrationTestUtils.copyConfigWithOverrides;
 import static java.util.Collections.singletonMap;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 
@@ -59,6 +60,7 @@ public class CassandraKVTableIntegrationTest {
   private static final int NUM_KAFKA_PARTITIONS = NUM_SUBPARTITIONS_TOTAL / 4;
 
   private CassandraClient client;
+  private CqlSession session;
   private String storeName; // ie the "kafkaName", NOT the "cassandraName"
   private ResponsiveConfig config;
 
@@ -68,7 +70,7 @@ public void before(
       final CassandraContainer<?> cassandra,
       @ResponsiveConfigParam final ResponsiveConfig config
   ) throws InterruptedException, TimeoutException {
-    final CqlSession session = CqlSession.builder()
+    this.session = CqlSession.builder()
         .addContactPoint(cassandra.getContactPoint())
         .withLocalDatacenter(cassandra.getLocalDatacenter())
         .withKeyspace("responsive_itests") // NOTE: this keyspace is expected to exist
@@ -196,8 +198,32 @@ public void shouldReturnRangeKeysInLexicalOrderAcrossMultipleSubPartitions() {
     }
   }
 
+  @SuppressWarnings("OptionalGetWithoutIsPresent")
   @Test
-  public void shouldRespectSemanticTtlForLookups() {
+  public void shouldRespectSemanticDefaultOnlyTtl()  {
+    // Given:
+    final var defaultTtl = Duration.ofMinutes(30);
+    final var ttlProvider = TtlProvider.<String, String>withDefault(defaultTtl);
+    final ResponsiveKeyValueParams params =
+        ResponsiveKeyValueParams.keyValue(storeName).withTtlProvider(ttlProvider);
+    final String tableName = params.name().tableName();
+
+    // When:
+    createTableFromParams(params);
+
+    // Then:
+    final var table = session.getMetadata()
+        .getKeyspace(session.getKeyspace().get())
+        .get()
+        .getTable(tableName)
+        .get();
+    final String describe = table.describe(false);
+
+    assertThat(describe, containsString("default_time_to_live = " + (int) defaultTtl.toSeconds()));
+  }
+
+  @Test
+  public void shouldRespectSemanticDefaultOnlyTtlForLookups() {
     // Given:
     final TtlProvider<?, ?> ttlProvider = TtlProvider.withDefault(Duration.ofMillis(100));
     final RemoteKVTable<BoundStatement> table = createTable(ttlProvider);