Support extract-key kafka message transform

runtime/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/composite/AsyncapiClientGenerator.java

@@ -19,10 +19,14 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Stream;
 
+import jakarta.json.bind.Jsonb;
+
 import io.aklivity.zilla.runtime.binding.asyncapi.config.AsyncapiSchemaConfig;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.config.AsyncapiBindingConfig;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.config.AsyncapiCompositeConditionConfig;
@@ -37,6 +41,7 @@
 import io.aklivity.zilla.runtime.binding.kafka.config.KafkaTopicConfig;
 import io.aklivity.zilla.runtime.binding.kafka.config.KafkaTopicConfigBuilder;
 import io.aklivity.zilla.runtime.binding.tcp.config.TcpOptionsConfig;
+import io.aklivity.zilla.runtime.catalog.inline.config.InlineOptionsConfigBuilder;
 import io.aklivity.zilla.runtime.engine.config.KindConfig;
 import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
 import io.aklivity.zilla.runtime.engine.config.NamespaceConfigBuilder;
@@ -89,7 +94,7 @@ private ClientNamespaceHelper(
             AsyncapiSchemaConfig schema)
         {
             super(config, schema.apiLabel);
-            this.catalogs = new CatalogsHelper(schema);
+            this.catalogs = new ClientCatalogsHelper(schema);
             this.bindings = new ClientBindingsHelper(schema);
         }
 
@@ -101,6 +106,37 @@ protected <C> NamespaceConfigBuilder<C> injectComponents(
                     .inject(bindings::injectAll);
         }
 
+        private final class ClientCatalogsHelper extends CatalogsHelper
+        {
+            private ClientCatalogsHelper(
+                AsyncapiSchemaConfig schema)
+            {
+                super(schema);
+            }
+
+            @Override
+            protected <C> void injectInlineSubject(
+                Jsonb jsonb,
+                InlineOptionsConfigBuilder<C> options,
+                AsyncapiMessageView message)
+            {
+                super.injectInlineSubject(jsonb, options, message);
+
+                if (message.bindings != null &&
+                    message.bindings.kafka != null &&
+                    message.bindings.kafka.key != null)
+                {
+                    final String subject = "%s-key".formatted(message.channel.address);
+
+                    options.schema()
+                        .subject(subject)
+                        .version("latest")
+                        .schema(toSchemaJson(jsonb, message.bindings.kafka.key.model))
+                        .build();
+                }
+            }
+        }
+
         private final class ClientBindingsHelper extends BindingsHelper
         {
             private static final Pattern PARAMETERIZED_TOPIC_PATTERN = Pattern.compile(REGEX_ADDRESS_PARAMETER);
@@ -142,7 +178,7 @@ private <C> NamespaceConfigBuilder<C> injectProtocols(
                     .map(s -> s.protocol)
                     .distinct()
                     .map(protocols::get)
-                    .filter(p -> p != null)
+                    .filter(Objects::nonNull)
                     .forEach(p -> p.inject(namespace));
 
                 return namespace;
@@ -152,10 +188,8 @@ private <C> NamespaceConfigBuilder<C> injectTlsClient(
                 NamespaceConfigBuilder<C> namespace)
             {
                 if (Stream.of(schema)
-                        .map(s -> s.asyncapi)
-                        .flatMap(v -> v.servers.stream())
-                        .filter(s -> secure.contains(s.protocol))
-                        .count() != 0L)
+                    .map(s -> s.asyncapi)
+                    .flatMap(v -> v.servers.stream()).anyMatch(s -> secure.contains(s.protocol)))
                 {
                     namespace
                         .binding()
@@ -204,6 +238,7 @@ private <C> NamespaceConfigBuilder<C> injectKafka(
                 NamespaceConfigBuilder<C> namespace)
             {
                 return namespace
+                    .inject(this::injectKafkaCache)
                     .binding()
                         .name("kafka_client0")
                         .type("kafka")
@@ -266,37 +301,41 @@ private <C> KafkaOptionsConfigBuilder<C> injectKafkaTopicOptions(
                 KafkaOptionsConfigBuilder<C> options)
             {
                 List<KafkaTopicConfig> topics = config.options.kafka != null
-                        ? config.options.kafka.topics
-                        : null;
+                    ? config.options.kafka.topics
+                    : null;
 
-                if (topics != null)
-                {
-                    Stream.of(schema)
-                        .map(s -> s.asyncapi)
-                        .flatMap(v -> v.operations.values().stream())
-                        .filter(o -> o.channel.hasMessages() || o.channel.hasParameters())
-                        .flatMap(o -> Stream.of(o.channel, o.reply != null ? o.reply.channel : null))
-                        .filter(c -> c != null)
-                        .forEach(channel ->
-                            topics.stream()
-                                .filter(t -> t.name.equals(channel.address))
-                                .findFirst()
-                                .ifPresent(topic ->
-                                    options
-                                        .topic()
-                                            .name(channel.address)
-                                            .transforms()
-                                                .extractHeaders(topic.transforms.extractHeaders)
-                                                .build()
-                                            .inject(t -> injectKafkaTopicKey(t, channel))
-                                            .inject(t -> injectKafkaTopicValue(t, channel))
-                                            .build()
-                                        .build()));
-                }
+                Stream.of(schema)
+                    .map(s -> s.asyncapi)
+                    .flatMap(v -> v.channels.values().stream())
+                    .filter(c -> !PARAMETERIZED_TOPIC_PATTERN.matcher(c.address).find())
+                    .distinct()
+                    .forEach(channel ->
+                        options.topic()
+                            .name(channel.address)
+                            .inject(t -> injectKafkaTopicTransforms(t, channel, topics))
+                            .inject(t -> injectKafkaTopicKey(t, channel))
+                            .inject(t -> injectKafkaTopicValue(t, channel))
+                            .build());
 
                 return options;
             }
 
+            private <C> KafkaTopicConfigBuilder<C> injectKafkaTopicTransforms(
+                KafkaTopicConfigBuilder<C> topic,
+                AsyncapiChannelView channel,
+                List<KafkaTopicConfig> topics)
+            {
+                Optional<KafkaTopicConfig> topicConfig = topics.stream()
+                    .filter(t -> t.name.equals(channel.address))
+                    .findFirst();
+                topicConfig.ifPresent(kafkaTopicConfig -> topic
+                            .transforms()
+                                .extractKey(kafkaTopicConfig.transforms.extractKey)
+                                .extractHeaders(kafkaTopicConfig.transforms.extractHeaders)
+                                .build());
+                return topic;
+            }
+
             private <C> KafkaTopicConfigBuilder<C> injectKafkaTopicKey(
                 KafkaTopicConfigBuilder<C> topic,
                 AsyncapiChannelView channel)
@@ -311,9 +350,10 @@ private <C> KafkaTopicConfigBuilder<C> injectKafkaTopicKey(
                                     .name("catalog0")
                                     .schema()
                                         .version("latest")
-                                        .subject(message.name)
+                                        .subject("%s-key".formatted(message.channel.address))
                                         .build()
-                                    .build());
+                                    .build()
+                                .build());
                 }
                 return topic;
             }
@@ -414,8 +454,7 @@ private <C> NamespaceConfigBuilder<C> injectSseClient(
                 if (Stream.of(schema)
                     .map(s -> s.asyncapi)
                     .flatMap(v -> v.operations.values().stream())
-                    .filter(AsyncapiOperationView::hasBindingsSse)
-                    .count() != 0L)
+                    .anyMatch(AsyncapiOperationView::hasBindingsSse))
                 {
                     namespace
                         .binding()

runtime/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/composite/AsyncapiCompositeGenerator.java

@@ -233,7 +233,7 @@ protected final String resolveIdentity(
             return value;
         }
 
-        protected final class CatalogsHelper
+        protected class CatalogsHelper
         {
             private final AsyncapiSchemaConfig schema;
 
@@ -353,39 +353,9 @@ private <C> InlineOptionsConfigBuilder<C> injectInlineSubjects(
                     Stream.of(schema)
                         .map(s -> s.asyncapi)
                         .flatMap(v -> v.operations.values().stream())
-                        .map(o -> o.channel)
-                        .filter(c -> c.messages != null)
-                        .flatMap(c -> c.messages.stream())
-                        .forEach(m ->
-                        {
-                            if (m.payload != null)
-                            {
-                                final String subject = "%s-%s-payload".formatted(m.channel.name, m.name);
-
-                                options.schema()
-                                    .subject(subject)
-                                    .version("latest")
-                                    .schema(toSchemaJson(jsonb, m.payload.model))
-                                    .build();
-                            }
-
-                            if (m.headers != null && m.headers.properties != null)
-                            {
-                                for (Map.Entry<String, AsyncapiSchemaView> header : m.headers.properties.entrySet())
-                                {
-                                    final String name = header.getKey();
-                                    final AsyncapiSchemaItemView schema = header.getValue();
-
-                                    final String subject = "%s-%s-header-%s".formatted(m.channel.name, m.name, name);
-
-                                    options.schema()
-                                        .subject(subject)
-                                        .version("latest")
-                                        .schema(toSchemaJson(jsonb, schema.model))
-                                        .build();
-                                }
-                            }
-                        });
+                        .filter(o -> o.messages != null)
+                        .flatMap(o -> o.messages.stream())
+                        .forEach(m -> injectInlineSubject(jsonb, options, m));
 
                     Stream.of(schema)
                         .map(s -> s.asyncapi)
@@ -413,7 +383,39 @@ private <C> InlineOptionsConfigBuilder<C> injectInlineSubjects(
                 return options;
             }
 
-            private static String toSchemaJson(
+            protected <C> void injectInlineSubject(
+                Jsonb jsonb,
+                InlineOptionsConfigBuilder<C> options,
+                AsyncapiMessageView message)
+            {
+                if (message.payload != null)
+                {
+                    options.schema()
+                        .subject("%s-%s-value".formatted(message.channel.name, message.name))
+                        .version("latest")
+                        .schema(toSchemaJson(jsonb, message.payload.model))
+                        .build();
+                }
+
+                if (message.headers != null && message.headers.properties != null)
+                {
+                    for (Map.Entry<String, AsyncapiSchemaView> header : message.headers.properties.entrySet())
+                    {
+                        final String name = header.getKey();
+                        final AsyncapiSchemaItemView schema = header.getValue();
+
+                        final String subject = "%s-header-%s".formatted(message.channel.address, name);
+
+                        options.schema()
+                            .subject(subject)
+                            .version("latest")
+                            .schema(toSchemaJson(jsonb, schema.model))
+                            .build();
+                    }
+                }
+            }
+
+            protected static String toSchemaJson(
                 Jsonb jsonb,
                 AsyncapiSchemaItem schema)
             {
@@ -473,7 +475,7 @@ protected final void injectPayloadModel(
                     message.contentType != null &&
                     modelContentType.reset(message.contentType).matches())
                 {
-                    final String subject = "%s-%s-payload".formatted(message.channel.name, message.name);
+                    final String subject = "%s-value".formatted(message.channel.address);
 
                     switch (modelContentType.group(1))
                     {

runtime/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/composite/AsyncapiProxyGenerator.java

@@ -634,7 +634,7 @@ private <C> BindingConfigBuilder<C> injectHttpKafkaRoute(
                         final String httpMethod = httpOperation.bindings.http.method;
                         final String httpPath = httpChannel.address;
 
-                        boolean async = httpChannel.messages.stream()
+                        boolean async = httpOperation.messages.stream()
                             .anyMatch(m -> m.correlationId != null);
 
                         if (async)
@@ -796,9 +796,7 @@ private <C> HttpKafkaWithProduceConfigBuilder<C> injectHttpKafkaRouteProduceWith
 
                     produce.acks("in_sync_replicas").key(key);
 
-                    AsyncapiChannelView httpChannel = httpOperation.channel;
-
-                    httpChannel.messages.forEach(message ->
+                    httpOperation.messages.forEach(message ->
                     {
                         if (message.correlationId != null)
                         {