Support Kafka consumer groups

incubator/command-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LoggableStream.java

@@ -87,6 +87,8 @@
 import io.aklivity.zilla.runtime.command.log.internal.types.stream.KafkaFetchDataExFW;
 import io.aklivity.zilla.runtime.command.log.internal.types.stream.KafkaFetchFlushExFW;
 import io.aklivity.zilla.runtime.command.log.internal.types.stream.KafkaFlushExFW;
+import io.aklivity.zilla.runtime.command.log.internal.types.stream.KafkaGroupBeginExFW;
+import io.aklivity.zilla.runtime.command.log.internal.types.stream.KafkaGroupDataExFW;
 import io.aklivity.zilla.runtime.command.log.internal.types.stream.KafkaMergedBeginExFW;
 import io.aklivity.zilla.runtime.command.log.internal.types.stream.KafkaMergedDataExFW;
 import io.aklivity.zilla.runtime.command.log.internal.types.stream.KafkaMergedFlushExFW;
@@ -890,6 +892,9 @@ private void onKafkaBeginEx(
         case KafkaBeginExFW.KIND_DESCRIBE:
             onKafkaDescribeBeginEx(offset, timestamp, kafkaBeginEx.describe());
             break;
+        case KafkaBeginExFW.KIND_GROUP:
+            onKafkaGroupBeginEx(offset, timestamp, kafkaBeginEx.group());
+            break;
         case KafkaBeginExFW.KIND_FETCH:
             onKafkaFetchBeginEx(offset, timestamp, kafkaBeginEx.fetch());
             break;
@@ -946,6 +951,19 @@ private void onKafkaDescribeBeginEx(
         configs.forEach(c -> out.printf(verboseFormat, index, offset, timestamp, c.asString()));
     }
 
+    private void onKafkaGroupBeginEx(
+        int offset,
+        long timestamp,
+        KafkaGroupBeginExFW group)
+    {
+        String16FW groupId = group.groupId();
+        String16FW protocol = group.protocol();
+        int timeout = group.timeout();
+
+        out.printf(verboseFormat, index, offset, timestamp, format("[group] %s %s %d",
+            groupId.asString(), protocol.asString(), timeout));
+    }
+
     private void onKafkaFetchBeginEx(
         int offset,
         long timestamp,
@@ -1062,6 +1080,9 @@ private void onKafkaDataEx(
         case KafkaDataExFW.KIND_DESCRIBE:
             onKafkaDescribeDataEx(offset, timestamp, kafkaDataEx.describe());
             break;
+        case KafkaDataExFW.KIND_GROUP:
+            onKafkaGroupDataEx(offset, timestamp, kafkaDataEx.group());
+            break;
         case KafkaDataExFW.KIND_FETCH:
             onKafkaFetchDataEx(offset, timestamp, kafkaDataEx.fetch());
             break;
@@ -1089,6 +1110,17 @@ private void onKafkaDescribeDataEx(
                                         format("%s: %s", c.name().asString(), c.value().asString())));
     }
 
+    private void onKafkaGroupDataEx(
+        int offset,
+        long timestamp,
+        KafkaGroupDataExFW group)
+    {
+        String16FW leader = group.leaderId();
+        String16FW member = group.memberId();
+
+        out.printf(verboseFormat, index, offset, timestamp, format("[group] %s %s", leader.asString(), member.asString()));
+    }
+
     private void onKafkaFetchDataEx(
         int offset,
         long timestamp,

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/KafkaConfiguration.java

@@ -23,6 +23,7 @@
 import java.math.BigInteger;
 import java.nio.file.Path;
 import java.security.SecureRandom;
+import java.util.UUID;
 import java.util.function.Supplier;
 
 import org.agrona.LangUtil;
@@ -63,6 +64,10 @@ public class KafkaConfiguration extends Configuration
     public static final IntPropertyDef KAFKA_CACHE_CLIENT_TRAILERS_SIZE_MAX;
     public static final IntPropertyDef KAFKA_CACHE_SERVER_RECONNECT_DELAY;
     public static final PropertyDef<NonceSupplier> KAFKA_CLIENT_SASL_SCRAM_NONCE;
+    public static final PropertyDef<String> KAFKA_CLIENT_GROUP_INSTANCE_ID;
+    public static final IntPropertyDef KAFKA_CLIENT_GROUP_REBALANCE_TIMEOUT;
+    public static final PropertyDef<String> KAFKA_CLIENT_ID;
+    public static final PropertyDef<InstanceIdSupplier> KAFKA_CLIENT_INSTANCE_ID_SUPPLIER;
 
     private static final ConfigurationDef KAFKA_CONFIG;
 
@@ -100,6 +105,11 @@ public class KafkaConfiguration extends Configuration
         KAFKA_CACHE_CLIENT_TRAILERS_SIZE_MAX = config.property("cache.client.trailers.size.max", 256);
         KAFKA_CLIENT_SASL_SCRAM_NONCE = config.property(NonceSupplier.class, "client.sasl.scram.nonce",
                 KafkaConfiguration::decodeNonceSupplier, KafkaConfiguration::defaultNonceSupplier);
+        KAFKA_CLIENT_GROUP_INSTANCE_ID = config.property("client.group.instance.id", UUID.randomUUID().toString());
+        KAFKA_CLIENT_GROUP_REBALANCE_TIMEOUT = config.property("client.group.rebalance.timeout", 4000);
+        KAFKA_CLIENT_ID = config.property("client.id", "zilla");
+        KAFKA_CLIENT_INSTANCE_ID_SUPPLIER = config.property(InstanceIdSupplier.class, "client.instance.id.supplier",
+            KafkaConfiguration::decodeInstanceIdSupplier, KafkaConfiguration::defaultInstanceIdSupplier);
         KAFKA_CONFIG = config;
     }
 
@@ -248,6 +258,21 @@ public int cacheClientTrailersSizeMax()
         return KAFKA_CACHE_CLIENT_TRAILERS_SIZE_MAX.getAsInt(this);
     }
 
+    public String clientGroupInstanceId()
+    {
+        return KAFKA_CLIENT_GROUP_INSTANCE_ID.get(this);
+    }
+
+    public String clientId()
+    {
+        return KAFKA_CLIENT_ID.get(this);
+    }
+
+    public int clientGroupRebalanceTimeout()
+    {
+        return KAFKA_CLIENT_GROUP_REBALANCE_TIMEOUT.getAsInt(this);
+    }
+
     private static Path cacheDirectory(
         Configuration config,
         String cacheDirectory)
@@ -267,6 +292,11 @@ public Supplier<String> nonceSupplier()
         return KAFKA_CLIENT_SASL_SCRAM_NONCE.get(this)::get;
     }
 
+    public Supplier<String> clientInstanceIdSupplier()
+    {
+        return KAFKA_CLIENT_INSTANCE_ID_SUPPLIER.get(this)::get;
+    }
+
     @FunctionalInterface
     private interface NonceSupplier
     {
@@ -315,4 +345,22 @@ private static NonceSupplier defaultNonceSupplier(
         return () ->
                  new BigInteger(130, new SecureRandom()).toString(Character.MAX_RADIX);
     }
+
+    @FunctionalInterface
+    private interface InstanceIdSupplier extends Supplier<String>
+    {
+    }
+
+    private static InstanceIdSupplier decodeInstanceIdSupplier(
+        Configuration config,
+        String value)
+    {
+        return () -> String.format("%s-%s", "zilla", UUID.randomUUID());
+    }
+
+    private static InstanceIdSupplier defaultInstanceIdSupplier(
+        Configuration config)
+    {
+        return () -> String.format("%s-%s", "zilla", UUID.randomUUID());
+    }
 }

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/config/KafkaBindingConfig.java

@@ -45,7 +45,18 @@ public KafkaRouteConfig resolve(
         String topic)
     {
         return routes.stream()
-            .filter(r -> r.authorized(authorization) && r.matches(topic))
+            .filter(r -> r.authorized(authorization) && r.matches(topic, null))
+            .findFirst()
+            .orElse(null);
+    }
+
+    public KafkaRouteConfig resolve(
+        long authorization,
+        String topic,
+        String groupId)
+    {
+        return routes.stream()
+            .filter(r -> r.authorized(authorization) && r.matches(topic, groupId))
             .findFirst()
             .orElse(null);
     }

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/config/KafkaConditionConfig.java

@@ -20,10 +20,13 @@
 public final class KafkaConditionConfig extends ConditionConfig
 {
     public final String topic;
+    public final String groupId;
 
     public KafkaConditionConfig(
-        String topic)
+        String topic,
+        String groupId)
     {
         this.topic = topic;
+        this.groupId = groupId;
     }
 }

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/config/KafkaConditionConfigAdapter.java

@@ -27,6 +27,7 @@
 public final class KafkaConditionConfigAdapter implements ConditionConfigAdapterSpi, JsonbAdapter<ConditionConfig, JsonObject>
 {
     private static final String TOPIC_NAME = "topic";
+    private static final String GROUP_ID_NAME = "groupId";
 
     @Override
     public String type()
@@ -47,6 +48,11 @@ public JsonObject adaptToJson(
             object.add(TOPIC_NAME, kafkaCondition.topic);
         }
 
+        if (kafkaCondition.groupId != null)
+        {
+            object.add(GROUP_ID_NAME, kafkaCondition.groupId);
+        }
+
         return object.build();
     }
 
@@ -58,6 +64,10 @@ public ConditionConfig adaptFromJson(
                 ? object.getString(TOPIC_NAME)
                 : null;
 
-        return new KafkaConditionConfig(topic);
+        String groupId = object.containsKey(GROUP_ID_NAME)
+                ? object.getString(GROUP_ID_NAME)
+                : null;
+
+        return new KafkaConditionConfig(topic, groupId);
     }
 }

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/config/KafkaConditionMatcher.java

@@ -21,17 +21,20 @@
 public final class KafkaConditionMatcher
 {
     private final Matcher topicMatch;
+    private final Matcher grouoIdMatch;
 
     public KafkaConditionMatcher(
         KafkaConditionConfig condition)
     {
         this.topicMatch = condition.topic != null ? asMatcher(condition.topic) : null;
+        this.grouoIdMatch = condition.groupId != null ? asMatcher(condition.groupId) : null;
     }
 
     public boolean matches(
-        String topic)
+        String topic,
+        String groupId)
     {
-        return matchesTopic(topic);
+        return matchesTopic(topic) && matchesGroupId(groupId);
     }
 
     private boolean matchesTopic(
@@ -40,9 +43,18 @@ private boolean matchesTopic(
         return this.topicMatch == null || this.topicMatch.reset(topic).matches();
     }
 
+    private boolean matchesGroupId(
+        String groupId)
+    {
+        return this.grouoIdMatch == null || this.grouoIdMatch.reset(groupId).matches();
+    }
+
     private static Matcher asMatcher(
         String wildcard)
     {
-        return Pattern.compile(wildcard.replace(".", "\\.").replace("*", ".*")).matcher("");
+        return Pattern.compile(wildcard
+            .replace(".", "\\.")
+            .replace("*", ".*"))
+            .matcher("");
     }
 }

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/config/KafkaRouteConfig.java

@@ -50,8 +50,9 @@ boolean authorized(
     }
 
     boolean matches(
-        String topic)
+        String topic,
+        String groupId)
     {
-        return when.isEmpty() || when.stream().anyMatch(m -> m.matches(topic));
+        return when.isEmpty() || when.stream().anyMatch(m -> m.matches(topic, groupId));
     }
 }

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientFactory.java

@@ -64,6 +64,8 @@ public KafkaCacheClientFactory(
         final KafkaCacheClientDescribeFactory cacheDescribeFactory = new KafkaCacheClientDescribeFactory(
                 config, context, bindings::get, supplyCacheRoute);
 
+        final KafkaCacheGroupFactory cacheGroupFactory = new KafkaCacheGroupFactory(config, context, bindings::get);
+
         final KafkaCacheClientFetchFactory cacheFetchFactory = new KafkaCacheClientFetchFactory(
                 config, context, bindings::get, accountant::supplyDebitor, supplyCache, supplyCacheRoute);
 
@@ -76,6 +78,7 @@ public KafkaCacheClientFactory(
         final Int2ObjectHashMap<BindingHandler> factories = new Int2ObjectHashMap<>();
         factories.put(KafkaBeginExFW.KIND_META, cacheMetaFactory);
         factories.put(KafkaBeginExFW.KIND_DESCRIBE, cacheDescribeFactory);
+        factories.put(KafkaBeginExFW.KIND_GROUP, cacheGroupFactory);
         factories.put(KafkaBeginExFW.KIND_FETCH, cacheFetchFactory);
         factories.put(KafkaBeginExFW.KIND_PRODUCE, cacheProduceFactory);
         factories.put(KafkaBeginExFW.KIND_MERGED, cacheMergedFactory);