kafkatopicsobserver: remove session_timeout and heartbeat_interval config

.chloggen/kafkatopicsobserver-rm-consumerconfig.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: breaking
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kafkatopicsobserver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Remove session_timeout and heartbeat_interval config
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38414]
+
+# (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: These attributes are only relevant to Kafka consumers.
+
+# 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]

extension/observer/kafkatopicsobserver/README.md

@@ -25,8 +25,6 @@ The following settings can be optionally configured:
 - `brokers` (default = localhost:9092): The list of kafka brokers
 - `resolve_canonical_bootstrap_servers_only` (default = false): Whether to resolve then reverse-lookup broker IPs during startup
 - `topic_regex` regex pattern of the topic name to subscribe to.
-- `session_timeout` (default = `10s`): The request timeout for detecting client failures when using Kafka’s group management facilities.
-- `heartbeat_interval` (default = `3s`): The expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities.
 - `auth`
     - `plain_text`
         - `username`: The username to use.
@@ -56,4 +54,4 @@ The following settings can be optionally configured:
         - `password`: The Kerberos password used for authenticate with KDC
         - `config_file`: Path to Kerberos configuration. i.e /etc/krb5.conf
         - `keytab_file`: Path to keytab file. i.e /etc/security/kafka.keytab
-        - `disable_fast_negotiation`: Disable PA-FX-FAST negotiation (Pre-Authentication Framework - Fast). Some common Kerberos implementations do not support PA-FX-FAST negotiation. This is set to `false` by default.
+        - `disable_fast_negotiation`: Disable PA-FX-FAST negotiation (Pre-Authentication Framework - Fast). Some common Kerberos implementations do not support PA-FX-FAST negotiation. This is set to `false` by default.

extension/observer/kafkatopicsobserver/config.go

@@ -22,11 +22,7 @@ type Config struct {
 	// required in SASL environments.
 	ResolveCanonicalBootstrapServersOnly bool `mapstructure:"resolve_canonical_bootstrap_servers_only"`
 	// Kafka protocol version
-	ProtocolVersion string `mapstructure:"protocol_version"`
-	// Session interval for the Kafka consumer
-	SessionTimeout time.Duration `mapstructure:"session_timeout"`
-	// Heartbeat interval for the Kafka consumer
-	HeartbeatInterval  time.Duration        `mapstructure:"heartbeat_interval"`
+	ProtocolVersion    string               `mapstructure:"protocol_version"`
 	Authentication     kafka.Authentication `mapstructure:"auth"`
 	TopicRegex         string               `mapstructure:"topic_regex"`
 	TopicsSyncInterval time.Duration        `mapstructure:"topics_sync_interval"`
@@ -45,11 +41,5 @@ func (config *Config) Validate() (errs error) {
 	if config.TopicsSyncInterval <= 0 {
 		errs = multierr.Append(errs, fmt.Errorf("topics_sync_interval must be greater than 0"))
 	}
-	if config.SessionTimeout <= 0 {
-		errs = multierr.Append(errs, fmt.Errorf("session_timeout must be greater than 0"))
-	}
-	if config.HeartbeatInterval <= 0 {
-		errs = multierr.Append(errs, fmt.Errorf("heartbeat_interval must be greater than 0"))
-	}
 	return errs
 }

extension/observer/kafkatopicsobserver/config_test.go

@@ -40,8 +40,6 @@ func TestLoadConfig(t *testing.T) {
 				TopicRegex:                           "^topic[0-9]$",
 				TopicsSyncInterval:                   5 * time.Second,
 				ResolveCanonicalBootstrapServersOnly: false,
-				SessionTimeout:                       30 * time.Second,
-				HeartbeatInterval:                    20 * time.Second,
 				Authentication: kafka.Authentication{
 					PlainText: &kafka.PlainTextConfig{
 						Username: "fooUser",
@@ -70,15 +68,13 @@ func TestValidateConfig(t *testing.T) {
 		ProtocolVersion: "3.7.0",
 		TopicRegex:      "^test[0-9]$",
 	}
-	assert.Equal(t, "brokers list must be specified; topics_sync_interval must be greater than 0; session_timeout must be greater than 0; heartbeat_interval must be greater than 0", xconfmap.Validate(cfg).Error())
+	assert.Equal(t, "brokers list must be specified; topics_sync_interval must be greater than 0", xconfmap.Validate(cfg).Error())
 
 	cfg = &Config{
 		Brokers:            []string{"1.2.3.4:9092"},
 		ProtocolVersion:    "",
 		TopicRegex:         "^topic[0-9]$",
 		TopicsSyncInterval: 1 * time.Second,
-		SessionTimeout:     1 * time.Second,
-		HeartbeatInterval:  1 * time.Second,
 	}
 	assert.Equal(t, "protocol_version must be specified", xconfmap.Validate(cfg).Error())
 
@@ -87,8 +83,6 @@ func TestValidateConfig(t *testing.T) {
 		ProtocolVersion:    "3.7.0",
 		TopicRegex:         "",
 		TopicsSyncInterval: 1 * time.Second,
-		SessionTimeout:     1 * time.Second,
-		HeartbeatInterval:  1 * time.Second,
 	}
 	assert.Equal(t, "topic_regex must be specified", xconfmap.Validate(cfg).Error())
 
@@ -97,8 +91,6 @@ func TestValidateConfig(t *testing.T) {
 		ProtocolVersion:    "3.7.0",
 		TopicRegex:         "^topic[0-9]$",
 		TopicsSyncInterval: 0 * time.Second,
-		SessionTimeout:     1 * time.Second,
-		HeartbeatInterval:  1 * time.Second,
 	}
 	assert.Equal(t, "topics_sync_interval must be greater than 0", xconfmap.Validate(cfg).Error())
 
@@ -107,28 +99,6 @@ func TestValidateConfig(t *testing.T) {
 		ProtocolVersion:    "3.7.0",
 		TopicRegex:         "^topic[0-9]$",
 		TopicsSyncInterval: 1 * time.Second,
-		SessionTimeout:     0 * time.Second,
-		HeartbeatInterval:  1 * time.Second,
-	}
-	assert.Equal(t, "session_timeout must be greater than 0", xconfmap.Validate(cfg).Error())
-
-	cfg = &Config{
-		Brokers:            []string{"1.2.3.4:9092"},
-		ProtocolVersion:    "3.7.0",
-		TopicRegex:         "^topic[0-9]$",
-		TopicsSyncInterval: 1 * time.Second,
-		SessionTimeout:     1 * time.Second,
-		HeartbeatInterval:  0 * time.Second,
-	}
-	assert.Equal(t, "heartbeat_interval must be greater than 0", xconfmap.Validate(cfg).Error())
-
-	cfg = &Config{
-		Brokers:            []string{"1.2.3.4:9092"},
-		ProtocolVersion:    "3.7.0",
-		TopicRegex:         "^topic[0-9]$",
-		TopicsSyncInterval: 1 * time.Second,
-		SessionTimeout:     1 * time.Second,
-		HeartbeatInterval:  1 * time.Second,
 	}
 	assert.NoError(t, xconfmap.Validate(cfg))
 }

extension/observer/kafkatopicsobserver/extension.go

@@ -130,8 +130,6 @@ func (k *kafkaTopicsObserver) Shutdown(_ context.Context) error {
 
 var createKafkaClusterAdmin = func(ctx context.Context, config Config) (sarama.ClusterAdmin, error) {
 	saramaConfig := sarama.NewConfig()
-	saramaConfig.Consumer.Group.Session.Timeout = config.SessionTimeout
-	saramaConfig.Consumer.Group.Heartbeat.Interval = config.HeartbeatInterval
 
 	var err error
 	if config.ResolveCanonicalBootstrapServersOnly {

extension/observer/kafkatopicsobserver/factory.go

@@ -15,8 +15,6 @@ import (
 
 const (
 	defaultBroker             = "localhost:9092"
-	defaultSessionTimeout     = 10 * time.Second
-	defaultHeartbeatInterval  = 3 * time.Second
 	defaultTopicsSyncInterval = 5 * time.Second
 )
 
@@ -33,8 +31,6 @@ func NewFactory() extension.Factory {
 func createDefaultConfig() component.Config {
 	return &Config{
 		Brokers:            []string{defaultBroker},
-		SessionTimeout:     defaultSessionTimeout,
-		HeartbeatInterval:  defaultHeartbeatInterval,
 		TopicsSyncInterval: defaultTopicsSyncInterval,
 	}
 }

extension/observer/kafkatopicsobserver/testdata/config.yaml

@@ -4,9 +4,7 @@ kafkatopics_observer/all_settings:
   topic_regex: "^topic[0-9]$"
   topics_sync_interval: 5s
   resolve_canonical_bootstrap_servers_only: false
-  session_timeout: 30s
-  heartbeat_interval: 20s
   auth:
     plain_text:
       username: fooUser
-      password: fooPassword
+      password: fooPassword