DBZ-6493 restructured spec.sink to more extendable format

debezium · May 31, 2023 · 3f5adbc · 3f5adbc
1 parent c2598cc
commit 3f5adbc
Show file tree

Hide file tree

Showing 6 changed files with 156 additions and 124 deletions.
diff --git a/README.md b/README.md
@@ -23,38 +23,39 @@ kubectl create -f examples/postgres/ -n $NAMESPACE
 ## DebeziumServerSpec Reference
 ```yaml
 spec:
-    version: String 
-    image: String # exclusive with version
-    storage:
-      type: persistent | ephemeral  # enum
-      claimName: String # only valid and required for persistent
-    runtime:
-      env: EnvFromSource array # https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.23/#envfromsource-v1-core
-      volumes: Volume array # https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.23/#volume-v1-core
-    quarkus:
-      # quarkus properties 
-    format:
-      value:
-        type: String
-        # other format properties
-      key:
-        type: String
-      header:
-        type: String 
-    transforms:
-      - name: String
-        type: String
-        predicate: String
-        negate: Boolean
-        # other transformation properties
-    predicates: 
-      - name: String
-        type: String
-        # other preticate properties
-    sink:
+  version: String
+  image: String # exclusive with version
+  storage:
+    type: persistent | ephemeral  # enum
+    claimName: String # only valid and required for persistent
+  runtime:
+    env: EnvFromSource array # https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.23/#envfromsource-v1-core
+    volumes: Volume array # https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.23/#volume-v1-core
+  quarkus:
+  # quarkus properties 
+  format:
+    value:
       type: String
-      # other sink properties
-    source:
-      class: String
-      # other source connector properties
+      # other format properties
+    key:
+      type: String
+    header:
+      type: String
+  transforms:
+    - name: String
+      type: String
+      predicate: String
+      negate: Boolean
+      # other transformation properties
+  predicates:
+    - name: String
+      type: String
+      # other preticate properties
+  sink:
+    type: String
+    config:
+    # other sink properties
+  source:
+    class: String
+    # other source connector properties
 ```
diff --git a/examples/postgres/010_debezium-server-ephemeral.yml b/examples/postgres/010_debezium-server-ephemeral.yml
@@ -12,9 +12,10 @@ spec:
     kubernetes-config.secrets: postgresql-credentials
   sink:
     type: kafka
-    producer.bootstrap.servers: dbz-kafka-kafka-bootstrap:9092
-    producer.key.serializer: org.apache.kafka.common.serialization.StringSerializer
-    producer.value.serializer: org.apache.kafka.common.serialization.StringSerializer
+    config:
+      producer.bootstrap.servers: dbz-kafka-kafka-bootstrap:9092
+      producer.key.serializer: org.apache.kafka.common.serialization.StringSerializer
+      producer.value.serializer: org.apache.kafka.common.serialization.StringSerializer
   source:
     class: io.debezium.connector.postgresql.PostgresConnector
     tasks.max: 

diff --git a/k8/debeziumservers.debezium.io-v1.yml b/k8/debeziumservers.debezium.io-v1.yml
@@ -17,75 +17,7 @@ spec:
         properties:
           spec:
             properties:
-              transforms:
-                items:
-                  properties:
-                    negate:
-                      type: boolean
-                    predicate:
-                      type: string
-                    name:
-                      type: string
-                    type:
-                      type: string
-                    props:
-                      additionalProperties:
-                        type: object
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
-                  type: object
-                  x-kubernetes-preserve-unknown-fields: true
-                type: array
-              sink:
-                properties:
-                  type:
-                    type: string
-                  props:
-                    additionalProperties:
-                      type: object
-                    type: object
-                    x-kubernetes-preserve-unknown-fields: true
-                type: object
-                x-kubernetes-preserve-unknown-fields: true
-              predicates:
-                items:
-                  properties:
-                    name:
-                      type: string
-                    type:
-                      type: string
-                    props:
-                      additionalProperties:
-                        type: object
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
-                  type: object
-                  x-kubernetes-preserve-unknown-fields: true
-                type: array
-              storage:
-                properties:
-                  claimName:
-                    type: string
-                  type:
-                    enum:
-                    - persistent
-                    - ephemeral
-                    type: string
-                type: object
-              version:
-                type: string
-              image:
-                type: string
-              quarkus:
-                properties:
-                  props:
-                    additionalProperties:
-                      type: object
-                    type: object
-                    x-kubernetes-preserve-unknown-fields: true
-                type: object
-                x-kubernetes-preserve-unknown-fields: true
-              externalConfiguration:
+              runtime:
                 properties:
                   volumes:
                     items:
@@ -758,6 +690,77 @@ spec:
                       type: object
                     type: array
                 type: object
+              transforms:
+                items:
+                  properties:
+                    negate:
+                      type: boolean
+                    predicate:
+                      type: string
+                    name:
+                      type: string
+                    type:
+                      type: string
+                    props:
+                      additionalProperties:
+                        type: object
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+                  type: object
+                  x-kubernetes-preserve-unknown-fields: true
+                type: array
+              sink:
+                properties:
+                  type:
+                    type: string
+                  config:
+                    properties:
+                      props:
+                        additionalProperties:
+                          type: object
+                        type: object
+                        x-kubernetes-preserve-unknown-fields: true
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
+                type: object
+              predicates:
+                items:
+                  properties:
+                    name:
+                      type: string
+                    type:
+                      type: string
+                    props:
+                      additionalProperties:
+                        type: object
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+                  type: object
+                  x-kubernetes-preserve-unknown-fields: true
+                type: array
+              storage:
+                properties:
+                  claimName:
+                    type: string
+                  type:
+                    enum:
+                    - persistent
+                    - ephemeral
+                    type: string
+                type: object
+              version:
+                type: string
+              image:
+                type: string
+              quarkus:
+                properties:
+                  props:
+                    additionalProperties:
+                      type: object
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
+                type: object
+                x-kubernetes-preserve-unknown-fields: true
               source:
                 properties:
                   class:

diff --git a/src/main/java/io/debezium/operator/config/ConfigMapping.java b/src/main/java/io/debezium/operator/config/ConfigMapping.java
@@ -80,11 +80,6 @@ public void put(String key, Map<String, ?> props) {
         props.forEach((subKey, value) -> putInternal(value, key, subKey));
     }
 
-    public void put(String key, String valueKey, Map<String, ?> props) {
-        put(key, valueKey);
-        put(valueKey, props);
-    }
-
     public <T extends ConfigMappable> void put(String key, Collection<T> items, Function<T, String> nameExtractor) {
         items.stream()
                 .map(nameExtractor)

diff --git a/src/main/java/io/debezium/operator/model/ConfigProperties.java b/src/main/java/io/debezium/operator/model/ConfigProperties.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.operator.model;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonAnyGetter;
+import com.fasterxml.jackson.annotation.JsonAnySetter;
+
+import io.debezium.operator.config.ConfigMappable;
+import io.debezium.operator.config.ConfigMapping;
+
+public class ConfigProperties implements ConfigMappable {
+
+    private Map<String, Object> props = new HashMap<>(0);
+
+    @JsonAnyGetter
+    public Map<String, Object> getProps() {
+        return props;
+    }
+
+    @JsonAnySetter
+    public void setProps(String name, Object value) {
+        getProps().put(name, value);
+    }
+
+    @Override
+    public ConfigMapping asConfiguration() {
+        return ConfigMapping.from(props);
+    }
+}
diff --git a/src/main/java/io/debezium/operator/model/Sink.java b/src/main/java/io/debezium/operator/model/Sink.java
@@ -5,19 +5,17 @@
  */
 package io.debezium.operator.model;
 
-import java.util.HashMap;
-import java.util.Map;
-
-import com.fasterxml.jackson.annotation.JsonAnyGetter;
-import com.fasterxml.jackson.annotation.JsonAnySetter;
-
 import io.debezium.operator.config.ConfigMappable;
 import io.debezium.operator.config.ConfigMapping;
 
 public class Sink implements ConfigMappable {
 
     private String type;
-    private Map<String, Object> props = new HashMap<>(0);
+    private ConfigProperties config;
+
+    public Sink() {
+        this.config = new ConfigProperties();
+    }
 
     public String getType() {
         return type;
@@ -27,20 +25,19 @@ public void setType(String type) {
         this.type = type;
     }
 
-    @JsonAnyGetter
-    public Map<String, Object> getProps() {
-        return props;
+    public ConfigProperties getConfig() {
+        return config;
     }
 
-    @JsonAnySetter
-    public void setProps(String name, Object value) {
-        getProps().put(name, value);
+    public void setConfig(ConfigProperties config) {
+        this.config = config;
     }
 
     @Override
     public ConfigMapping asConfiguration() {
         var config = ConfigMapping.empty();
-        config.put("type", type, props);
+        config.put("type", type);
+        config.put(type, this.config);
         return config;
     }
 }