Plural broker field in pipeline config (#278)

config.yaml

@@ -1,5 +1,5 @@
 environment: development
-broker: "http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092"
+brokers: "http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092"
 kafka_connect_host: "http://localhost:8083"
 kafka_rest_host: "http://localhost:8082"
 schema_registry_url: "http://localhost:8081"

docs/docs/resources/pipeline-components/kafka-app.yaml

@@ -7,7 +7,7 @@
   # add the key-value pairs they need.
   app: # required
     streams: # required
-      brokers: ${broker} # required
+      brokers: ${brokers} # required
       schemaRegistryUrl: ${schema_registry_url}
     nameOverride: override-with-this-name # kafka-app-specific
     imageTag: "1.0.0" # Example values that are shared between streams-app and producer-app 

docs/docs/resources/pipeline-components/pipeline.yaml

@@ -50,7 +50,7 @@
   # add the key-value pairs they need.
   app: # required
     streams: # required
-      brokers: ${broker} # required
+      brokers: ${brokers} # required
       schemaRegistryUrl: ${schema_registry_url}
     nameOverride: override-with-this-name # kafka-app-specific
     imageTag: "1.0.0" # Example values that are shared between streams-app and producer-app
@@ -96,7 +96,7 @@
   app: # required
     # Streams Bootstrap streams section
     streams: # required, streams-app-specific
-      brokers: ${broker} # required
+      brokers: ${brokers} # required
       schemaRegistryUrl: ${schema_registry_url}
       inputTopics:
         - topic1
@@ -190,7 +190,7 @@
   # https://github.com/bakdata/streams-bootstrap/tree/master/charts/producer-app
   app: # required
     streams: # required, producer-specific
-      brokers: ${broker} # required
+      brokers: ${brokers} # required
       schemaRegistryUrl: ${schema_registry_url}
       outputTopic: output_topic
       extraOutputTopics: 

docs/docs/resources/pipeline-components/producer.yaml

@@ -9,7 +9,7 @@
   # https://github.com/bakdata/streams-bootstrap/tree/master/charts/producer-app
   app: # required
     streams: # required, producer-specific
-      brokers: ${broker} # required
+      brokers: ${brokers} # required
       schemaRegistryUrl: ${schema_registry_url}
       outputTopic: output_topic
       extraOutputTopics: 

docs/docs/resources/pipeline-components/streams-app.yaml

@@ -9,7 +9,7 @@
   app: # required
     # Streams Bootstrap streams section
     streams: # required, streams-app-specific
-      brokers: ${broker} # required
+      brokers: ${brokers} # required
       schemaRegistryUrl: ${schema_registry_url}
       inputTopics:
         - topic1

docs/docs/resources/pipeline-defaults/defaults-kafka-app.yaml

@@ -5,6 +5,6 @@
 kafka-app:
   app:
     streams:
-      brokers: ${broker}
+      brokers: ${brokers}
       schemaRegistryUrl: ${schema_registry_url}
     nameOverride: override-with-this-name

docs/docs/resources/pipeline-defaults/defaults.yaml

@@ -52,7 +52,7 @@ kubernetes-app:
 kafka-app:
   app:
     streams:
-      brokers: ${broker}
+      brokers: ${brokers}
       schemaRegistryUrl: ${schema_registry_url}
     nameOverride: override-with-this-name
 #

docs/docs/resources/pipeline_config/config.yaml

@@ -8,9 +8,9 @@ defaults_path: .
 # pipeline_development.yaml for environment=development).
 # REQUIRED
 environment: development
-# The Kafka broker address.
+# The Kafka brokers address.
 # REQUIRED
-broker: "http://localhost:9092"
+brokers: "http://broker1:9092,http://broker2:9092"
 # The name of the defaults file and the prefix of the defaults environment file.
 defaults_filename_prefix: defaults
 # Configures topic names.

docs/docs/schema/config.json

@@ -39,13 +39,14 @@
             "additionalProperties": false,
             "description": "Pipeline configuration unrelated to the components.",
             "properties": {
-                "broker": {
-                    "description": "The Kafka broker address.",
-                    "env": "KPOPS_KAFKA_BROKER",
+                "brokers": {
+                    "description": "The comma separated Kafka brokers address.",
+                    "env": "KPOPS_KAFKA_BROKERS",
                     "env_names": [
-                        "kpops_kafka_broker"
+                        "kpops_kafka_brokers"
                     ],
-                    "title": "Broker",
+                    "example": "broker1:9092,broker2:9092,broker3:9092",
+                    "title": "Brokers",
                     "type": "string"
                 },
                 "create_namespace": {
@@ -187,7 +188,7 @@
             },
             "required": [
                 "environment",
-                "broker"
+                "brokers"
             ],
             "title": "PipelineConfig",
             "type": "object"

examples/bakdata/atm-fraud-detection/config.yaml

@@ -4,7 +4,7 @@ topic_name_config:
   default_error_topic_name: "${pipeline_name}-${component_name}-dead-letter-topic"
   default_output_topic_name: "${pipeline_name}-${component_name}-topic"
 
-broker: "http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092"
+brokers: "http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092"
 
 schema_registry_url: "http://k8kafka-cp-schema-registry.kpops.svc.cluster.local:8081"
 

examples/bakdata/atm-fraud-detection/defaults.yaml

@@ -10,7 +10,7 @@ kafka-connector:
 kafka-app:
   app:
     streams:
-      brokers: ${broker}
+      brokers: ${brokers}
       schemaRegistryUrl: ${schema_registry_url}
       optimizeLeaveGroupBehavior: false
 

kpops/cli/pipeline_config.py

@@ -38,10 +38,11 @@ class PipelineConfig(BaseSettings):
         description="The environment you want to generate and deploy the pipeline to. "
         "Suffix your environment files with this value (e.g. defaults_development.yaml for environment=development).",
     )
-    broker: str = Field(
+    brokers: str = Field(
         default=...,
-        env=f"{ENV_PREFIX}KAFKA_BROKER",
-        description="The Kafka broker address.",
+        env=f"{ENV_PREFIX}KAFKA_BROKERS",
+        description="The comma separated Kafka brokers address.",
+        example="broker1:9092,broker2:9092,broker3:9092",
     )
     defaults_filename_prefix: str = Field(
         default="defaults",

kpops/components/base_components/kafka_connector.py

@@ -271,7 +271,7 @@ def _get_kafka_connect_resetter_values(
             **KafkaConnectResetterValues(
                 config=KafkaConnectResetterConfig(
                     connector=connector_name,
-                    brokers=self.config.broker,
+                    brokers=self.config.brokers,
                     **kwargs,
                 ),
                 connector_type=connector_type.value,

tests/components/test_kafka_sink_connector.py

@@ -48,7 +48,7 @@ def config(self) -> PipelineConfig:
                 default_error_topic_name="${component_type}-error-topic",
                 default_output_topic_name="${component_type}-output-topic",
             ),
-            broker="broker:9092",
+            brokers="broker:9092",
             helm_diff_config=HelmDiffConfig(),
         )
 

tests/components/test_kafka_source_connector.py

@@ -41,7 +41,7 @@ def config(slef) -> PipelineConfig:
                 default_error_topic_name="${component_type}-error-topic",
                 default_output_topic_name="${component_type}-output-topic",
             ),
-            broker="broker:9092",
+            brokers="broker:9092",
             helm_diff_config=HelmDiffConfig(),
         )
 

tests/pipeline/resources/custom-config/config.yaml

@@ -3,7 +3,7 @@ defaults_path: ../no-topics-defaults
 topic_name_config:
   default_error_topic_name: "${component_name}-dead-letter-topic"
   default_output_topic_name: "${component_name}-test-topic"
-broker: "http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092"
+brokers: "http://k8kafka-cp-kafka-headless.kpops.svc.cluster.local:9092"
 kafka_connect_host: "http://localhost:8083"
 kafka_rest_host: "http://localhost:8082"
 schema_registry_url: "http://localhost:8081"

tests/pipeline/resources/defaults.yaml

@@ -5,7 +5,7 @@ kubernetes-app:
 kafka-app:
   app:
     streams:
-      brokers: "${broker}"
+      brokers: "${brokers}"
       schema_registry_url: "${schema_registry_url}"
   version: "2.4.2"
 

tests/pipeline/resources/kafka-connect-sink-config/config.yaml

@@ -1,6 +1,6 @@
 environment: development
 defaults_path: ..
-broker: "broker:9092"
+brokers: "broker:9092"
 topic_name_config: 
   default_error_topic_name: ${component_type}-error-topic
   default_output_topic_name: ${component_type}-output-topic

tests/pipeline/resources/no-topics-defaults/defaults.yaml

@@ -1,7 +1,7 @@
 kafka-app:
   app:
     streams:
-      brokers: "${broker}"
+      brokers: "${brokers}"
       schemaRegistryUrl: "${schema_registry_url}"
 
 producer:

tests/pipeline/resources/pipeline-component-should-have-prefix/defaults.yaml

@@ -7,5 +7,5 @@ kubernetes-app:
 kafka-app:
   app:
     streams:
-      brokers: ${broker}
+      brokers: ${brokers}
       schemaRegistryUrl: ${schema_registry_url}

tests/pipeline/resources/pipeline-with-env-defaults/defaults.yaml

@@ -4,7 +4,7 @@ kubernetes-app:
 kafka-app:
   app:
     streams:
-      brokers: "${broker}"
+      brokers: "${brokers}"
       schemaRegistryUrl: "${schema_registry_url}"
 
 producer: {} # inherits from kafka