ingest-storage: support multiple kafka seed broker addresses

[srpvpn]

Summary

Add support for multiple Kafka seed brokers via comma-separated values in ingest-storage.kafka.address.

Changes

• Extended KafkaConfig.SeedBrokers() to parse and trim comma-separated broker addresses
• Updated Kafka client initialization to use kgo.SeedBrokers(cfg.SeedBrokers()...)
• Added unit tests for broker parsing and CreateTopic with multiple brokers
• Updated documentation to reflect CSV format support

Testing

go test ./pkg/storage/ingest -run TestKafkaConfig_SeedBrokers -v
go test ./pkg/storage/ingest -run 'TestCreateTopic/should_support_multiple_comma-separated_seed_broker_addresses' -v
go test ./pkg/storage/ingest -run 'TestConfig_(Validate|GetConsumerGroup)|TestKafkaConfig_SeedBrokers|TestCreateTopic' -v

Fix #14303

---

Note

Medium Risk
Touches Kafka connection/bootstrap configuration and its parsing/validation, which can impact ingest availability if misconfigured. Changes are scoped and covered by updated tests and docs.

Overview
Adds support for configuring multiple Kafka seed brokers for ingest storage by changing ingest-storage.kafka.address from a single string to a CSV-backed list and wiring it through Kafka client creation (kgo.SeedBrokers(cfg.Address...)).

Updates validation, tests, fixture generation, and various ingest/distributor test setups to use the new list type, and refreshes CLI help + docs + changelog to document the comma-separated format.

^{Written by Cursor Bugbot for commit 8009b06. This will update automatically on new commits. Configure here.}

[CLAassistant]

All committers have signed the CLA.

[56quarters]

Thanks for the PR! I'm not super familiar with ingest storage so I'll let someone from the ingest squad take a look.

[56quarters]

There's a type in dskit called StringSliceCSV that handles this. There's an example here.

[srpvpn]

Thanks, I fixed it

[narqo]

I think, the comment above was about using the flagext.StringSliceCSV for cfg.Address. Any reason we can't do that here?

[srpvpn]

Okay: KafkaConfig.Address now uses flagext.StringSliceCSV directly, and the flag is registered via f.Var(&cfg.Address, ...). SeedBrokers still keeps trimming and
backward-compatible fallback behavior. Also updated ingest tests/fixtures to set address through the same CSV parsing path.

[tacole02]

Docs look good! Thank you!

[narqo]

Thank you for iterating on that. Please see the comments. After addressing those, please check the tests. I think there are more places around Mimir, that use KafkaConfig.Address. Those tests need to be fixed.

[narqo]

This test is redundant and doesn't bring value: it checks what must be guarantied by the type used. Please, cut.

[narqo]

Can you explain what this subtest has to do with TestCreateTopic — does passing multiple seed addresses change how a topic is created?

[srpvpn]

Removed the redundant TestKafkaConfig_AddressCSV and dropped the unrelated multi-seed subtest from TestCreateTopic.

[narqo]

I think there are more places around Mimir, that use KafkaConfig.Address. Those tests need to be fixed.

Judging by the broken CI, this comment still needs to be address.

[srpvpn]

@narqo Addressed. I fixed the remaining test usages of KafkaConfig.Address

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Bugbot Autofix is ON, but it could not run because the branch was deleted or merged before Autofix could start.}

[cursor]

Broker addresses not trimmed, spaces cause connection failures

Medium Severity

flagext.StringSliceCSV.Set splits on commas without trimming whitespace from individual items. A common user input like "broker1:9092, broker2:9092" produces ["broker1:9092", " broker2:9092"] — note the leading space. The franz-go parseBrokerAddr function does not trim whitespace either, so the untrimmed address broker2:9092 causes DNS resolution failures. No validation or trimming is performed in Validate() or anywhere in the pipeline to catch this.

Additional Locations (1)

• pkg/storage/ingest/util.go#L69-L70