fix(#4533): support multiple Kafka bootstrap servers

[stefaniesinner]

Purpose

Fixes #4533. So far the Kafka adapter and sink took one hostname and one port, so if that single broker was down, the connection failed even though the rest of the cluster was fine.

Both now have one Kafka Brokers field taking a comma-separated list like broker1:9092,broker2:9092. Kafka only needs one of them to answer and finds the rest from there. KafkaBootstrapServersParser checks the notation and strips whitespace, empty entries, and duplicates.

Adapter and sink go to version 3. Existing ones keep working: their old host and port get joined into the new field automatically.

Remarks

• In KafkaProtocol.resolveConfiguration I moved the config extraction into the try block. Otherwise a bad broker just shows up as a generic error instead of a message at the field.
• KafkaAdapterTester reads the config list by index, so merging two fields into one shifted everything after the broker and I had to adjust those.

PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no

[Copilot]

Pull request overview

This PR updates the Kafka adapter and Kafka sink configuration to accept a comma-separated bootstrap.servers list (instead of a single host+port), adds parsing/validation + migrations to keep existing stored configurations working, and adapts tests/docs accordingly.

Changes:

• Replace Kafka host/port fields with a single bootstrap-servers field across adapter + sink (incl. migrations to v3).
• Propagate the broker list through the Kafka messaging layer (KafkaTransportProtocol + config factories + producer).
• Add/adjust unit, integration, and UI e2e tests plus documentation strings.

Reviewed changes

Copilot reviewed 25 out of 25 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
ui/cypress/tests/thirdparty/Kafka.smoke.spec.ts	Update UI smoke test input field
streampipes-model/src/main/java/org/apache/streampipes/model/grounding/KafkaTransportProtocol.java	Add bootstrap servers to protocol model
streampipes-messaging-kafka/src/test/java/org/apache/streampipes/messaging/kafka/config/ProducerConfigFactoryTest.java	New tests for bootstrap propagation
streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaProducer.java	Use bootstrap servers in producer
streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/config/AbstractConfigFactory.java	Resolve broker URL via bootstrap list
streampipes-messaging-kafka/pom.xml	Add JUnit API test dependency
streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/KafkaAdapterTester.java	Adjust integration test template config
streampipes-extensions/streampipes-connectors-kafka/src/test/java/org/apache/streampipes/extensions/connectors/kafka/sink/KafkaPublishSinkTest.java	New sink config extraction tests
streampipes-extensions/streampipes-connectors-kafka/src/test/java/org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaBootstrapServersParserTest.java	New parser tests
streampipes-extensions/streampipes-connectors-kafka/src/test/java/org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaBootstrapServersMergerTest.java	New migration merge tests
streampipes-extensions/streampipes-connectors-kafka/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.kafka/strings.en	Update sink i18n labels
streampipes-extensions/streampipes-connectors-kafka/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.kafka/documentation.md	Document multi-broker sink config
streampipes-extensions/streampipes-connectors-kafka/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.kafka/strings.en	Update adapter i18n labels
streampipes-extensions/streampipes-connectors-kafka/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.kafka/documentation.md	Document multi-broker adapter config
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/sink/KafkaPublishSink.java	Use bootstrap servers for sink producer
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaConfigProvider.java	Replace host/port keys with bootstrap key
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaConfigExtractor.java	Parse/validate bootstrap server list
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaBootstrapServersParser.java	New bootstrap servers parser
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaBootstrapServersMerger.java	New legacy host/port → bootstrap merger
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaBaseConfig.java	Replace host/port with bootstrap string
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/migration/KafkaSinkMigrationV3.java	Migrate sink config v2→v3
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/migration/KafkaAdapterMigrationV3.java	Migrate adapter config v2→v3
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/KafkaConnectorsModuleExport.java	Register new v3 migrators
streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/adapter/KafkaProtocol.java	Use bootstrap servers throughout adapter
streampipes-extensions/streampipes-connectors-kafka/pom.xml	Add JUnit test dependencies

Suppressed comments (1)

streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaProducer.java:74

• The 3-arg SpKafkaProducer constructor creates a local KafkaTransportProtocol but does not assign it to the instance field protocol. If an instance created via this constructor is ever disconnected and reconnected (or connect() is called for any reason), connect() will dereference protocol and throw an NPE. Assigning the created protocol to this.protocol keeps the object state consistent and avoids this failure mode.

  public SpKafkaProducer(String url,
                         String topic,
                         List<KafkaConfigAppender> appenders) {
    KafkaTransportProtocol protocol = new KafkaTransportProtocol();
    protocol.setBootstrapServers(url);
    protocol.setTopicDefinition(new SimpleTopicDefinition(topic));
    this.brokerUrl = url;
    this.topic = topic;
    this.producer = new KafkaProducer<>(makeProperties(protocol, appenders));
    this.connected = true;

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.