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

Kafka 4.0 removed DescribeConfigs API v0 (the broker minimum is now v1),
so the hardcoded v0 request used for topic and broker configuration
collection failed against 4.x brokers. As a result KafkaTopicSample was
dropped entirely, while KafkaBrokerSample/KafkaProducerSample/
KafkaConsumerSample/KafkaOffsetSample (JMX- and other-API-based) kept
working - matching the reported "topic samples missing on Kafka 4.x".

Derive the DescribeConfigs request version from the negotiated
KAFKA_VERSION (v2 if >= 2.0.0, v1 if >= 1.1.0, else v0; sarama v1.43.x
supports DescribeConfigs up to v2), and raise the default KAFKA_VERSION
from 1.0.0 to 2.1.0 so topic and broker configuration collection works
against Kafka 4.x out of the box with no user configuration.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: shashank-reddy-nr

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## Unreleased
 
+### 🐛 Bug fixes
+- 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,