fix: collect KafkaTopicSample on Kafka 4.x (DescribeConfigs version)

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## Unreleased
 
+### Bugfixes
+- Fixed missing `KafkaTopicSample` data on Kafka 4.x. The integration sent `DescribeConfigs` requests at API version 0, which Kafka 4.0 removed (the broker minimum is now v1), causing topic configuration collection to fail and the topic samples to be dropped entirely. The request version is now derived from the configured `KAFKA_VERSION` (v2/v1/v0), and the default `KAFKA_VERSION` was raised from `1.0.0` to `2.1.0` so topic and broker configuration collection works against Kafka 4.x out of the box.
+
 ## v3.20.0 - 2026-06-17
 
 ### 🛡️ Security notices

src/args/args.go

@@ -10,7 +10,7 @@ import (
 type ArgumentList struct {
 	sdkArgs.DefaultArgumentList
 	ClusterName  string `default:"" help:"A user-defined name to uniquely identify the cluster"`
-	KafkaVersion string `default:"1.0.0" help:"What version of the API to use for connecting to Kafka. Versions older than 1.0.0 may be missing some features."`
+	KafkaVersion string `default:"2.1.0" help:"What version of the API to use for connecting to Kafka. Defaults to 2.1.0. Set this to match your cluster's version. Kafka 4.x requires at least 2.0.0 for topic and broker configuration collection (DescribeConfigs v0 was removed). Versions older than 1.0.0 may be missing some features."`
 
 	AutodiscoverStrategy string `default:"zookeeper" help:"How to discover broker nodes to collect metrics from. One of 'zookeeper' (default) or 'bootstrap'"`
 

src/args/args_test.go

@@ -126,7 +126,7 @@ func TestDefaultArgs(t *testing.T) {
 			CacheTTL:  persist.DefaultTTL,
 		},
 		AutodiscoverStrategy: "zookeeper",
-		KafkaVersion:         sarama.V1_0_0_0,
+		KafkaVersion:         sarama.V2_1_0_0,
 		ZookeeperHosts: []*ZookeeperHost{
 			{
 				Host: "localhost",

src/broker/broker_collection.go

@@ -191,9 +191,20 @@ func collectBrokerTopicMetrics(b *connection.Broker, collectedTopics []string, i
 // Collect broker configuration from Zookeeper
 func getBrokerConfig(broker *connection.Broker) ([]*sarama.ConfigEntry, error) {
 
+	// DescribeConfigs v0 was removed in Kafka 4.0 (the broker's minimum supported
+	// version is now v1). Derive the request version from the negotiated Kafka
+	// version so we remain compatible with both older brokers and Kafka 4.x.
+	// sarama v1.43.x supports DescribeConfigs up to v2.
+	descVer := int16(0)
+	switch {
+	case args.GlobalArgs.KafkaVersion.IsAtLeast(sarama.V2_0_0_0):
+		descVer = 2
+	case args.GlobalArgs.KafkaVersion.IsAtLeast(sarama.V1_1_0_0):
+		descVer = 1
+	}
 	configRequest := &sarama.DescribeConfigsRequest{
-		Version:         0,
-		IncludeSynonyms: true,
+		Version:         descVer,
+		IncludeSynonyms: descVer >= 1,
 		Resources: []*sarama.ConfigResource{
 			{
 				Type:        sarama.BrokerResource,

src/topic/topic_collection.go

@@ -206,9 +206,20 @@ func topicRespondsToMetadata(t *Topic, client connection.Client) int {
 
 // Collect and populate the remainder of the topic struct fields
 func setTopicInfo(t *Topic, client connection.Client) error {
+	// DescribeConfigs v0 was removed in Kafka 4.0 (the broker's minimum supported
+	// version is now v1). Derive the request version from the negotiated Kafka
+	// version so we remain compatible with both older brokers and Kafka 4.x.
+	// sarama v1.43.x supports DescribeConfigs up to v2.
+	descVer := int16(0)
+	switch {
+	case args.GlobalArgs.KafkaVersion.IsAtLeast(sarama.V2_0_0_0):
+		descVer = 2
+	case args.GlobalArgs.KafkaVersion.IsAtLeast(sarama.V1_1_0_0):
+		descVer = 1
+	}
 	configRequest := &sarama.DescribeConfigsRequest{
-		Version:         0,
-		IncludeSynonyms: true,
+		Version:         descVer,
+		IncludeSynonyms: descVer >= 1,
 		Resources: []*sarama.ConfigResource{
 			{
 				Type:        sarama.TopicResource,