Applied feedback from PR

Author: akrambek

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/schema/pgsql.kafka.schema.patch.json

@@ -36,6 +36,7 @@
                     "routes": false,
                     "required":
                     [
+                        "catalog",
                         "exit"
                     ]
                 }

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/PgsqlKafkaConfiguration.java

@@ -31,7 +31,7 @@ public class PgsqlKafkaConfiguration extends Configuration
         KAFKA_TOPIC_REQUEST_TIMEOUT_MS = config.property("kafka.topic.request.timeout.ms", 30000);
         KAFKA_CREATE_TOPICS_PARTITION_COUNT = config.property("kafka.create.topics.partition.count", 1);
         KAFKA_CREATE_TOPICS_REPLICAS = config.property("kafka.create.topics.replicas", (short) 1);
-        KAFKA_AVRO_SCHEMA_NAMESPACE = config.property("kafka.avro.schema.namespace", "io.aklivity.zilla");
+        KAFKA_AVRO_SCHEMA_NAMESPACE = config.property("kafka.avro.schema.namespace", "{database}");
         PGSQL_KAFKA_CONFIG = config;
     }
 

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/schema/PgsqlKafkaValueAvroSchemaTemplate.java

@@ -22,6 +22,8 @@
 
 public class PgsqlKafkaValueAvroSchemaTemplate extends PgsqlKafkaAvroSchemaTemplate
 {
+    private static final String DATABASE_PLACEHOLDER = "{database}";
+
     private final StringBuilder schemaBuilder = new StringBuilder();
     private final String namespace;
 
@@ -37,12 +39,14 @@ public String generateSchema(
     {
         schemaBuilder.setLength(0);
 
+        final String newNamespace = namespace.replace(DATABASE_PLACEHOLDER, database);
+
         final String recordName = String.format("%s.%s", database, createTable.getTable().getName());
 
         schemaBuilder.append("{\n");
         schemaBuilder.append("\"type\": \"record\",\n");
         schemaBuilder.append("\"name\": \"").append(recordName).append("\",\n");
-        schemaBuilder.append("\"namespace\": \"").append(namespace).append("\",\n");
+        schemaBuilder.append("\"namespace\": \"").append(newNamespace).append("\",\n");
         schemaBuilder.append("\"fields\": [\n");
 
         for (ColumnDefinition column : createTable.getColumnDefinitions())

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/PgsqlKafkaCommandType.java

@@ -15,6 +15,9 @@
 package io.aklivity.zilla.runtime.binding.pgsql.kafka.internal.stream;
 
 import java.util.Arrays;
+import java.util.Map;
+
+import org.agrona.collections.Object2ObjectHashMap;
 
 public enum PgsqlKafkaCommandType
 {
@@ -23,6 +26,16 @@ public enum PgsqlKafkaCommandType
 
     private final byte[] value;
 
+    private static final Map<String, PgsqlKafkaCommandType> COMMAND_MAP = new Object2ObjectHashMap<>();
+
+    static
+    {
+        for (PgsqlKafkaCommandType commandType : PgsqlKafkaCommandType.values())
+        {
+            COMMAND_MAP.put(Arrays.toString(commandType.value), commandType);
+        }
+    }
+
     PgsqlKafkaCommandType(byte[] value)
     {
         this.value = value;
@@ -33,21 +46,8 @@ public byte[] value()
         return value;
     }
 
-    public static PgsqlKafkaCommandType valueOf(
-        byte[] value)
+    public static PgsqlKafkaCommandType valueOf(byte[] value)
     {
-        PgsqlKafkaCommandType command = UNKNOWN_COMMAND;
-
-        command:
-        for (PgsqlKafkaCommandType commandType : PgsqlKafkaCommandType.values())
-        {
-            if (Arrays.equals(commandType.value, value))
-            {
-                command = commandType;
-                break command;
-            }
-        }
-
-        return command;
+        return COMMAND_MAP.getOrDefault(Arrays.toString(value), UNKNOWN_COMMAND);
     }
 }

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/PgsqlKafkaProxyFactory.java

@@ -135,8 +135,8 @@ public final class PgsqlKafkaProxyFactory implements PgsqlKafkaStreamFactory
     {
         Object2ObjectHashMap<PgsqlKafkaCommandType, PgsqlDecoder> pgsqlDecoder =
             new Object2ObjectHashMap<>();
-        pgsqlDecoder.put(PgsqlKafkaCommandType.CREATE_TOPIC_COMMAND, this::onDecodeCreateTopicCommand);
-        pgsqlDecoder.put(PgsqlKafkaCommandType.UNKNOWN_COMMAND, this::onDecodeUnknownCommand);
+        pgsqlDecoder.put(PgsqlKafkaCommandType.CREATE_TOPIC_COMMAND, this::decodeCreateTopicCommand);
+        pgsqlDecoder.put(PgsqlKafkaCommandType.UNKNOWN_COMMAND, this::decodeUnknownCommand);
         this.pgsqlDecoder = pgsqlDecoder;
     }
 
@@ -1319,7 +1319,7 @@ private MessageConsumer newKafkaConsumer(
         return receiver;
     }
 
-    private void onDecodeCreateTopicCommand(
+    private void decodeCreateTopicCommand(
         PgsqlProxy server,
         long traceId,
         long authorization,
@@ -1357,7 +1357,7 @@ else if (server.commandsProcessed == 1)
         }
     }
 
-    private void onDecodeUnknownCommand(
+    private void decodeUnknownCommand(
         PgsqlProxy server,
         long traceId,
         long authorization,