minor global handling update

responsivedev · Oct 29, 2024 · 21c692e · 21c692e
1 parent 5bf5978
commit 21c692e
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 4 deletions.
diff --git a/kafka-client/src/main/java/dev/responsive/kafka/internal/stores/GlobalOperations.java b/kafka-client/src/main/java/dev/responsive/kafka/internal/stores/GlobalOperations.java
@@ -50,10 +50,11 @@ public class GlobalOperations implements KeyValueOperations {
 
   public static GlobalOperations create(
       final StateStoreContext storeContext,
-      final ResponsiveKeyValueParams params
+      final ResponsiveKeyValueParams params,
+      final Optional<TtlResolver<?, ?>> ttlResolver
   ) throws InterruptedException, TimeoutException {
 
-    if (params.ttlProvider().isPresent()) {
+    if (ttlResolver.isPresent()) {
       throw new UnsupportedOperationException("Global stores are not yet compatible with ttl");
     }
 
@@ -68,7 +69,7 @@ public static GlobalOperations create(
     final var spec = RemoteTableSpecFactory.fromKVParams(
         params,
         defaultPartitioner(),
-        Optional.empty()
+        ttlResolver
     );
 
     final var table = client.globalFactory().create(spec);

diff --git a/kafka-client/src/main/java/dev/responsive/kafka/internal/stores/ResponsiveKeyValueStore.java b/kafka-client/src/main/java/dev/responsive/kafka/internal/stores/ResponsiveKeyValueStore.java
@@ -141,7 +141,7 @@ private static KeyValueOperations provideOperations(
       final TaskType taskType
   ) throws InterruptedException, TimeoutException {
     return (taskType == TaskType.GLOBAL)
-        ? GlobalOperations.create(context, params)
+        ? GlobalOperations.create(context, params, ttlResolver)
         : PartitionedOperations.create(params.name(), ttlResolver, context, params);
   }