KIP-924: Custom assignment integration test

buildSrc/build.gradle.kts

@@ -19,6 +19,7 @@ plugins {
 }
 
 repositories {
+    mavenLocal()
     gradlePluginPortal()
 }
 

buildSrc/src/main/kotlin/responsive.java-common-conventions.gradle.kts

@@ -44,6 +44,7 @@ checkstyle {
 
 repositories {
     mavenCentral()
+    mavenLocal()
 }
 
 tasks.test {

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

@@ -34,6 +34,7 @@
 import dev.responsive.kafka.api.async.internals.AsyncThreadPoolRegistry;
 import dev.responsive.kafka.api.config.CompatibilityMode;
 import dev.responsive.kafka.api.config.ResponsiveConfig;
+import dev.responsive.kafka.api.stores.ResponsiveDslStoreSuppliers;
 import dev.responsive.kafka.internal.clients.ResponsiveKafkaClientSupplier;
 import dev.responsive.kafka.internal.config.ConfigUtils;
 import dev.responsive.kafka.internal.config.InternalSessionConfigs;
@@ -271,7 +272,6 @@ private static Properties propsWithOverrides(
   ) {
     final Properties propsWithOverrides = new Properties();
 
-
     final InternalSessionConfigs.Builder internalConfBuilder = new InternalSessionConfigs.Builder()
         .withSessionClients(sessionClients)
         .withStoreRegistry(storeRegistry)
@@ -313,6 +313,11 @@ private static Properties propsWithOverrides(
       throw new ConfigException(errorMsg);
     }
 
+    propsWithOverrides.putIfAbsent(
+        StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG,
+        ResponsiveDslStoreSuppliers.class.getName()
+    );
+
     return propsWithOverrides;
   }
 

kafka-client/src/main/java/dev/responsive/kafka/api/stores/ResponsiveDslStoreSuppliers.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2023 Responsive Computing, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dev.responsive.kafka.api.stores;
+
+import org.apache.kafka.streams.state.DslKeyValueParams;
+import org.apache.kafka.streams.state.DslSessionParams;
+import org.apache.kafka.streams.state.DslStoreSuppliers;
+import org.apache.kafka.streams.state.DslWindowParams;
+import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.state.SessionBytesStoreSupplier;
+import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
+
+public class ResponsiveDslStoreSuppliers implements DslStoreSuppliers {
+
+  @Override
+  public KeyValueBytesStoreSupplier keyValueStore(final DslKeyValueParams dslKeyValueParams) {
+    return ResponsiveStores.keyValueStore(
+        ResponsiveKeyValueParams.keyValue(dslKeyValueParams.name())
+    );
+  }
+
+  @Override
+  public WindowBytesStoreSupplier windowStore(final DslWindowParams dslWindowParams) {
+    return ResponsiveStores.windowStoreSupplier(
+        ResponsiveWindowParams.window(
+            dslWindowParams.name(),
+            dslWindowParams.windowSize(),
+            dslWindowParams.retentionPeriod().minus(dslWindowParams.windowSize())
+        )
+    );
+  }
+
+  @Override
+  public SessionBytesStoreSupplier sessionStore(final DslSessionParams dslSessionParams) {
+    return ResponsiveStores.sessionStoreSupplier(
+        ResponsiveSessionParams.session(
+            dslSessionParams.name(),
+            dslSessionParams.retentionPeriod()
+        )
+    );
+  }
+}

kafka-client/src/main/java/dev/responsive/kafka/api/stores/ResponsiveSessionParams.java

@@ -28,35 +28,36 @@ public final class ResponsiveSessionParams {
 
   private final TableName name;
   private final SessionSchema schemaType;
-  private final long inactivityGapMs;
-  private final long gracePeriodMs;
   private final long retentionPeriodMs;
-
-  private long numSegments;
+  private final long numSegments;
 
   private ResponsiveSessionParams(
       final String name,
       final SessionSchema schemaType,
-      final Duration inactivityGap,
-      final Duration gracePeriod
+      final long retentionPeriodMs
   ) {
     this.name = new TableName(name);
     this.schemaType = schemaType;
-    this.inactivityGapMs = durationToMillis(inactivityGap, "inactivityGap");
-    this.gracePeriodMs = durationToMillis(gracePeriod, "gracePeriod");
-
-    this.retentionPeriodMs = this.inactivityGapMs + this.gracePeriodMs;
+    this.retentionPeriodMs = retentionPeriodMs;
     this.numSegments = computeDefaultNumSegments(retentionPeriodMs);
   }
 
+  public static ResponsiveSessionParams session(
+      final String name,
+      final Duration retention
+  ) {
+    return new ResponsiveSessionParams(name, SessionSchema.SESSION, retention.toMillis());
+  }
+
   public static ResponsiveSessionParams session(
       final String name,
       final Duration inactivityGap,
       final Duration gracePeriod
   ) {
-    return new ResponsiveSessionParams(
-        name, SessionSchema.SESSION, inactivityGap, gracePeriod
-    );
+    final long inactivityGapMs = durationToMillis(inactivityGap, "inactivityGap");
+    final long gracePeriodMs = durationToMillis(gracePeriod, "gracePeriod");
+    final long retentionPeriodMs = inactivityGapMs + gracePeriodMs;
+    return new ResponsiveSessionParams(name, SessionSchema.SESSION, retentionPeriodMs);
   }
 
   public SessionSchema schemaType() {
@@ -75,10 +76,6 @@ public long numSegments() {
     return this.numSegments;
   }
 
-  public long gracePeriodMs() {
-    return this.gracePeriodMs;
-  }
-
   private static long computeDefaultNumSegments(final long retentionPeriodMs) {
     // TODO: Smart implementation.
     return DEFAULT_NUM_SEGMENTS;

kafka-client/src/main/java/dev/responsive/kafka/api/stores/ResponsiveStores.java

@@ -316,6 +316,19 @@ public static <K, V> Materialized<K, V, WindowStore<Bytes, byte[]>> windowMateri
     );
   }
 
+  /**
+   * See for example {@link Stores#inMemorySessionStore(String, Duration)}
+   *
+   * @param params the {@link ResponsiveSessionParams} for this store
+   * @return a supplier for a session store with the given options
+   *         that uses Responsive's storage for its backend
+   */
+  public static SessionBytesStoreSupplier sessionStoreSupplier(
+      final ResponsiveSessionParams params
+  ) {
+    return new ResponsiveSessionStoreSupplier(params);
+  }
+
   /**
    * Create a {@link StoreBuilder} that can be used to build a Responsive
    * {@link SessionStore} and connect it via the Processor API. If using the DSL, use

kafka-client/src/main/java/dev/responsive/kafka/internal/clients/DelegatingConsumer.java

@@ -34,6 +34,7 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 
 public abstract class DelegatingConsumer<K, V> implements Consumer<K, V> {
 
@@ -292,4 +293,9 @@ public void close(final Duration timeout) {
   public void wakeup() {
     delegate.wakeup();
   }
+
+  @Override
+  public Uuid clientInstanceId(final Duration duration) {
+    return delegate.clientInstanceId(duration);
+  }
 }

kafka-client/src/main/java/dev/responsive/kafka/internal/clients/ResponsiveProducer.java

@@ -34,6 +34,7 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -146,6 +147,11 @@ public List<PartitionInfo> partitionsFor(final String topic) {
     return wrapped.metrics();
   }
 
+  @Override
+  public Uuid clientInstanceId(final Duration duration) {
+    return wrapped.clientInstanceId(duration);
+  }
+
   @Override
   public void close() {
     wrapped.close();