Support parameters in KafkaTopicsConfig

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

@@ -18,10 +18,9 @@
 import static io.aklivity.zilla.runtime.binding.kafka.internal.types.KafkaOffsetType.HISTORICAL;
 import static java.util.stream.Collectors.toList;
 
+import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.function.ToLongFunction;
-import java.util.stream.Collectors;
 
 import io.aklivity.zilla.runtime.binding.kafka.config.KafkaOptionsConfig;
 import io.aklivity.zilla.runtime.binding.kafka.config.KafkaSaslConfig;
@@ -32,20 +31,18 @@
 import io.aklivity.zilla.runtime.engine.EngineContext;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
 import io.aklivity.zilla.runtime.engine.config.KindConfig;
-import io.aklivity.zilla.runtime.engine.model.ConverterHandler;
 
 public final class KafkaBindingConfig
 {
+    private static final KafkaTopicType DEFAULT_TOPIC_TYPE = new KafkaTopicType();
+
     public final long id;
     public final String name;
     public final KafkaOptionsConfig options;
     public final KindConfig kind;
     public final List<KafkaRouteConfig> routes;
     public final ToLongFunction<String> resolveId;
-    public final Map<String, ConverterHandler> keyReaders;
-    public final Map<String, ConverterHandler> keyWriters;
-    public final Map<String, ConverterHandler> valueReaders;
-    public final Map<String, ConverterHandler> valueWriters;
+    public final List<KafkaTopicType> kafkaTopicTypes;
 
     public KafkaBindingConfig(
         BindingConfig binding,
@@ -57,38 +54,8 @@ public KafkaBindingConfig(
         this.options = KafkaOptionsConfig.class.cast(binding.options);
         this.routes = binding.routes.stream().map(KafkaRouteConfig::new).collect(toList());
         this.resolveId = binding.resolveId;
-        this.keyReaders = options != null && options.topics != null
-                ? options.topics.stream()
-                .collect(Collectors.toMap(
-                    t -> t.name,
-                    t -> t.key != null
-                        ? context.supplyReadConverter(t.key)
-                        : ConverterHandler.NONE))
-                : null;
-        this.keyWriters = options != null && options.topics != null
-                ? options.topics.stream()
-                .collect(Collectors.toMap(
-                    t -> t.name,
-                    t -> t.key != null
-                        ? context.supplyWriteConverter(t.key)
-                        : ConverterHandler.NONE))
-                : null;
-        this.valueReaders = options != null && options.topics != null
-                ? options.topics.stream()
-                .collect(Collectors.toMap(
-                    t -> t.name,
-                    t -> t.value != null
-                        ? context.supplyReadConverter(t.value)
-                        : ConverterHandler.NONE))
-                : null;
-        this.valueWriters = options != null && options.topics != null
-                ? options.topics.stream()
-                .collect(Collectors.toMap(
-                    t -> t.name,
-                    t -> t.value != null
-                        ? context.supplyWriteConverter(t.value)
-                        : ConverterHandler.NONE))
-                : null;
+        this.kafkaTopicTypes = options != null && options.topics != null
+            ? options.topics.stream().map(t -> new KafkaTopicType(context, t)).collect(toList()) : Collections.emptyList();
     }
 
     public KafkaRouteConfig resolve(
@@ -147,27 +114,18 @@ public KafkaOffsetType supplyDefaultOffset(
         return config != null && config.defaultOffset != null ? config.defaultOffset : HISTORICAL;
     }
 
-    public ConverterHandler resolveKeyReader(
-        String topic)
-    {
-        return keyReaders != null ? keyReaders.getOrDefault(topic, ConverterHandler.NONE) : ConverterHandler.NONE;
-    }
-
-    public ConverterHandler resolveKeyWriter(
-        String topic)
-    {
-        return keyWriters != null ? keyWriters.getOrDefault(topic, ConverterHandler.NONE) : ConverterHandler.NONE;
-    }
-
-    public ConverterHandler resolveValueReader(
-        String topic)
-    {
-        return valueReaders != null ? valueReaders.getOrDefault(topic, ConverterHandler.NONE) : ConverterHandler.NONE;
-    }
-
-    public ConverterHandler resolveValueWriter(
+    public KafkaTopicType resolveTopicType(
         String topic)
     {
-        return valueWriters != null ? valueWriters.getOrDefault(topic, ConverterHandler.NONE) : ConverterHandler.NONE;
+        KafkaTopicType topicType = DEFAULT_TOPIC_TYPE;
+        for (KafkaTopicType k : kafkaTopicTypes)
+        {
+            if (k.matches(topic))
+            {
+                topicType = k;
+                break;
+            }
+        }
+        return topicType;
     }
 }

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc.
+ *
+ * Aklivity licenses this file to you 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 io.aklivity.zilla.runtime.binding.kafka.internal.config;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import io.aklivity.zilla.runtime.binding.kafka.config.KafkaTopicConfig;
+import io.aklivity.zilla.runtime.engine.EngineContext;
+import io.aklivity.zilla.runtime.engine.model.ConverterHandler;
+
+public class KafkaTopicType
+{
+    private final Matcher topicMatch;
+    public final ConverterHandler keyReader;
+    public final ConverterHandler keyWriter;
+    public final ConverterHandler valueReader;
+    public final ConverterHandler valueWriter;
+
+    public KafkaTopicType()
+    {
+        this.topicMatch = null;
+        this.keyReader = ConverterHandler.NONE;
+        this.keyWriter = ConverterHandler.NONE;
+        this.valueReader = ConverterHandler.NONE;
+        this.valueWriter = ConverterHandler.NONE;
+    }
+
+    public KafkaTopicType(
+        EngineContext context,
+        KafkaTopicConfig topicConfig)
+    {
+        this.topicMatch = topicConfig.name != null ? asMatcher(topicConfig.name) : null;
+        this.keyReader = topicConfig.key != null ? context.supplyReadConverter(topicConfig.key) : ConverterHandler.NONE;
+        this.keyWriter = topicConfig.key != null ? context.supplyWriteConverter(topicConfig.key) : ConverterHandler.NONE;
+        this.valueReader = topicConfig.value != null ? context.supplyReadConverter(topicConfig.value) : ConverterHandler.NONE;
+        this.valueWriter = topicConfig.value != null ? context.supplyWriteConverter(topicConfig.value) : ConverterHandler.NONE;
+    }
+
+    public boolean matches(
+        String topic)
+    {
+        return this.topicMatch == null || this.topicMatch.reset(topic).matches();
+    }
+
+    private static Matcher asMatcher(
+        String topic)
+    {
+        return Pattern.compile(topic.replaceAll("\\{[^}]+\\}", "[^/]+")).matcher("");
+    }
+}

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

@@ -48,6 +48,7 @@
 import io.aklivity.zilla.runtime.binding.kafka.internal.cache.KafkaCacheTopic;
 import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaBindingConfig;
 import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaRouteConfig;
+import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaTopicType;
 import io.aklivity.zilla.runtime.binding.kafka.internal.types.Array32FW;
 import io.aklivity.zilla.runtime.binding.kafka.internal.types.Flyweight;
 import io.aklivity.zilla.runtime.binding.kafka.internal.types.KafkaAckMode;
@@ -260,11 +261,10 @@ public MessageConsumer newStream(
                 final KafkaCache cache = supplyCache.apply(cacheName);
                 final KafkaCacheTopic topic = cache.supplyTopic(topicName);
                 final KafkaCachePartition partition = topic.supplyProducePartition(partitionId, localIndex);
-                final ConverterHandler convertKey = binding.resolveKeyWriter(topicName);
-                final ConverterHandler convertValue = binding.resolveValueWriter(topicName);
+                final KafkaTopicType topicType = binding.resolveTopicType(topicName);
                 final KafkaCacheClientProduceFan newFan =
                         new KafkaCacheClientProduceFan(routedId, resolvedId, authorization, budget,
-                            partition, cacheRoute, topicName, convertKey, convertValue);
+                            partition, cacheRoute, topicName, topicType);
 
                 cacheRoute.clientProduceFansByTopicPartition.put(partitionKey, newFan);
                 fan = newFan;
@@ -537,8 +537,7 @@ private KafkaCacheClientProduceFan(
             KafkaCachePartition partition,
             KafkaCacheRoute cacheRoute,
             String topicName,
-            ConverterHandler convertKey,
-            ConverterHandler convertValue)
+            KafkaTopicType topicType)
         {
             this.originId = originId;
             this.routedId = routedId;
@@ -548,8 +547,8 @@ private KafkaCacheClientProduceFan(
             this.budget = budget;
             this.cacheRoute = cacheRoute;
             this.topicName = topicName;
-            this.convertKey = convertKey;
-            this.convertValue = convertValue;
+            this.convertKey = topicType.keyWriter;
+            this.convertValue = topicType.valueWriter;
             this.members = new Long2ObjectHashMap<>();
             this.defaultOffset = KafkaOffsetType.LIVE;
             this.cursor = cursorFactory.newCursor(

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

@@ -52,6 +52,7 @@
 import io.aklivity.zilla.runtime.binding.kafka.internal.cache.KafkaCacheTopic;
 import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaBindingConfig;
 import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaRouteConfig;
+import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaTopicType;
 import io.aklivity.zilla.runtime.binding.kafka.internal.types.Array32FW;
 import io.aklivity.zilla.runtime.binding.kafka.internal.types.ArrayFW;
 import io.aklivity.zilla.runtime.binding.kafka.internal.types.Flyweight;
@@ -237,11 +238,10 @@ public MessageConsumer newStream(
                 final KafkaCache cache = supplyCache.apply(cacheName);
                 final KafkaCacheTopic cacheTopic = cache.supplyTopic(topicName);
                 final KafkaCachePartition partition = cacheTopic.supplyFetchPartition(partitionId);
-                final ConverterHandler convertKey = binding.resolveKeyReader(topicName);
-                final ConverterHandler convertValue = binding.resolveValueReader(topicName);
+                final KafkaTopicType topicType = binding.resolveTopicType(topicName);
                 final KafkaCacheServerFetchFanout newFanout =
                     new KafkaCacheServerFetchFanout(routedId, resolvedId, authorization,
-                        affinity, partition, routeDeltaType, defaultOffset, convertKey, convertValue);
+                        affinity, partition, routeDeltaType, defaultOffset, topicType);
 
                 cacheRoute.serverFetchFanoutsByTopicPartition.put(partitionKey, newFanout);
                 fanout = newFanout;
@@ -516,8 +516,7 @@ private KafkaCacheServerFetchFanout(
             KafkaCachePartition partition,
             KafkaDeltaType deltaType,
             KafkaOffsetType defaultOffset,
-            ConverterHandler convertKey,
-            ConverterHandler convertValue)
+            KafkaTopicType topicType)
         {
             this.originId = originId;
             this.routedId = routedId;
@@ -528,8 +527,8 @@ private KafkaCacheServerFetchFanout(
             this.retentionMillisMax = defaultOffset == LIVE ? SECONDS.toMillis(30) : Long.MAX_VALUE;
             this.members = new ArrayList<>();
             this.leaderId = leaderId;
-            this.convertKey = convertKey;
-            this.convertValue = convertValue;
+            this.convertKey = topicType.keyReader;
+            this.convertValue = topicType.valueReader;
             this.entryMark = new MutableInteger(0);
             this.valueMark = new MutableInteger(0);
         }