[receiver/kafka_metrics_receiver] Add refresh_frequency to enable custom refreshing for cluster metadata (#37897)

Fixes:
#37896

There are other ways to solve this issue, but that would require changes
in Sarama library.
This PR proposes a solution to exposes `refresh_frequency` from Sarama
library and let the user configure it.

Today the default setting from Sarama is being used which defaults to 10
mins.

More info and similar issues are mentioned in the tracking issue.

Added unit tests, and tested manually with a 3 brokers kafka-kraft
cluster.

---------

Signed-off-by: Shivanshu Raj Shrivastava <[email protected]>
Co-authored-by: Dmitrii Anoshin <[email protected]>

Author: shivanshuraj1333

.chloggen/kafka_metrics_receiver-add-refresh_frequency.yaml

@@ -0,0 +1,29 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: receiver/kafkametricsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `refresh_frequency` config to `kafkametricsreceiver`, to configure custom duration for cluster metadata refresh
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37896]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  - Helps fine tuning the refresh_frequency, and enables custom cluster metadata refresh intervals
+  - Default refresh_frequency is set 10 minutes from Sarama library defaults
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/kafkametricsreceiver/config.go

@@ -4,6 +4,8 @@
 package kafkametricsreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkametricsreceiver"
 
 import (
+	"time"
+
 	"go.opentelemetry.io/collector/scraper/scraperhelper"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka"
@@ -38,6 +40,11 @@ type Config struct {
 	// Authentication data
 	Authentication kafka.Authentication `mapstructure:"auth"`
 
+	// Cluster metadata refresh frequency
+	// Configures the refresh frequency to update cached cluster metadata
+	// Defaults to 10 minutes from Sarama library
+	RefreshFrequency time.Duration `mapstructure:"refresh_frequency"`
+
 	// Scrapers defines which metric data points to be captured from kafka
 	Scrapers []string `mapstructure:"scrapers"`
 

receiver/kafkametricsreceiver/config_test.go

@@ -44,6 +44,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 			},
 		},
+		RefreshFrequency:     1,
 		ClientID:             defaultClientID,
 		Scrapers:             []string{"brokers", "topics", "consumers"},
 		MetricsBuilderConfig: metadata.DefaultMetricsBuilderConfig(),

receiver/kafkametricsreceiver/receiver.go

@@ -49,6 +49,11 @@ var newMetricsReceiver = func(
 		}
 		sc.Version = version
 	}
+
+	if config.RefreshFrequency != 0 {
+		sc.Metadata.RefreshFrequency = config.RefreshFrequency
+	}
+
 	if err := kafka.ConfigureAuthentication(ctx, config.Authentication, sc); err != nil {
 		return nil, err
 	}

receiver/kafkametricsreceiver/receiver_test.go

@@ -59,6 +59,14 @@ func TestNewReceiver_invalid_auth_error(t *testing.T) {
 	assert.Nil(t, r)
 }
 
+func TestNewReceiver_refresh_frequency(t *testing.T) {
+	c := createDefaultConfig().(*Config)
+	c.RefreshFrequency = 1
+	r, err := newMetricsReceiver(context.Background(), *c, receivertest.NewNopSettings(), nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, r)
+}
+
 func TestNewReceiver(t *testing.T) {
 	c := createDefaultConfig().(*Config)
 	c.Scrapers = []string{"brokers"}

receiver/kafkametricsreceiver/testdata/config.yaml

@@ -11,5 +11,6 @@ kafkametrics:
       ca_file: ca.pem
       cert_file: cert.pem
       key_file: key.pem
+  refresh_frequency: 1
   topic_match: test_\w+
   group_match: test_\w+