Kafka GRPC consumer Group Support

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

@@ -14,15 +14,25 @@
  */
 package io.aklivity.zilla.runtime.binding.kafka.grpc.internal;
 
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.util.function.Supplier;
+
+import org.agrona.LangUtil;
+
 import io.aklivity.zilla.runtime.engine.Configuration;
 
 public class KafkaGrpcConfiguration extends Configuration
 {
     private static final ConfigurationDef KAFKA_GRPC_CONFIG;
+    public static final PropertyDef<GroupIdSupplier> KAFKA_GROUP_ID;
 
     static
     {
         final ConfigurationDef config = new ConfigurationDef("zilla.binding.kafka.grpc");
+        KAFKA_GROUP_ID = config.property(GroupIdSupplier.class, "group.id",
+            KafkaGrpcConfiguration::decodeGroupId, KafkaGrpcConfiguration::defaultGroupId);
         KAFKA_GRPC_CONFIG = config;
     }
 
@@ -31,4 +41,56 @@ public KafkaGrpcConfiguration(
     {
         super(KAFKA_GRPC_CONFIG, config);
     }
+
+    public Supplier<String> groupIdSupplier()
+    {
+        return KAFKA_GROUP_ID.get(this)::get;
+    }
+
+    @FunctionalInterface
+    private interface GroupIdSupplier extends Supplier<String>
+    {
+    }
+
+    private static GroupIdSupplier decodeGroupId(
+        Configuration config,
+        String fullyQualifiedMethodName)
+    {
+        GroupIdSupplier supplier = null;
+
+        try
+        {
+            MethodType signature = MethodType.methodType(String.class);
+            String[] parts = fullyQualifiedMethodName.split("::");
+            Class<?> ownerClass = Class.forName(parts[0]);
+            String methodName = parts[1];
+            MethodHandle method = MethodHandles.publicLookup().findStatic(ownerClass, methodName, signature);
+            supplier = () ->
+            {
+                String value = null;
+                try
+                {
+                    value = (String) method.invoke();
+                }
+                catch (Throwable ex)
+                {
+                    LangUtil.rethrowUnchecked(ex);
+                }
+
+                return value;
+            };
+        }
+        catch (Throwable ex)
+        {
+            LangUtil.rethrowUnchecked(ex);
+        }
+
+        return supplier;
+    }
+
+    private static GroupIdSupplier defaultGroupId(
+        Configuration config)
+    {
+        return () -> String.format("zilla-%s-%s", "", "");
+    }
 }

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

@@ -21,6 +21,7 @@
 import io.aklivity.zilla.runtime.binding.kafka.grpc.config.KafkaGrpcCorrelationConfig;
 import io.aklivity.zilla.runtime.binding.kafka.grpc.internal.types.Array32FW;
 import io.aklivity.zilla.runtime.binding.kafka.grpc.internal.types.KafkaHeaderFW;
+import io.aklivity.zilla.runtime.binding.kafka.grpc.internal.types.KafkaOffsetFW;
 import io.aklivity.zilla.runtime.binding.kafka.grpc.internal.types.OctetsFW;
 import io.aklivity.zilla.runtime.binding.kafka.grpc.internal.types.stream.KafkaDataExFW;
 import io.aklivity.zilla.runtime.binding.kafka.grpc.internal.types.stream.KafkaMergedFetchDataExFW;
@@ -32,6 +33,8 @@ public final class KafkaGrpcFetchHeaderHelper
     private final OctetsFW methodRO = new OctetsFW();
     private final OctetsFW replyToRO = new OctetsFW();
     private final OctetsFW correlatedIdRO = new OctetsFW();
+    public int partitionId;
+    public long partitionOffset;
 
     public OctetsFW service;
     public OctetsFW method;
@@ -65,6 +68,11 @@ public void visit(
         {
             final KafkaMergedFetchDataExFW kafkaMergedFetchDataEx = dataEx.merged().fetch();
             final Array32FW<KafkaHeaderFW> headers = kafkaMergedFetchDataEx.headers();
+            final KafkaOffsetFW partition = kafkaMergedFetchDataEx.partition();
+
+            partitionId = partition.partitionId();
+            partitionOffset = partition.partitionOffset();
+
             headers.forEach(this::dispatch);
         }
     }