[receiver/kafka] Remove deprecated legacy topic and encoding (open-telemetry#44568)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
This PR removes the deprecated topic and encoding config


<!--Please delete paragraphs that you did not use before submitting.-->

Author: khushijain21

.chloggen/remove-deprecated-topic.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: receiver/kafka
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Remove deprecated topic and encoding
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [44568]
+
+# (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:
+
+# 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/kafkareceiver/config.go

@@ -34,24 +34,6 @@ type Config struct {
 	// Profiles holds configuration about how profiles should be consumed.
 	Profiles TopicEncodingConfig `mapstructure:"profiles"`
 
-	// Topic holds the name of the Kafka topic from which to consume data.
-	//
-	// Topic has no default. If explicitly specified, it will take precedence
-	// over the default values of Logs.Topic, Traces.Topic, and Metrics.Topic.
-	//
-	// Deprecated [v0.124.0]: Use Logs.Topic, Traces.Topic, and Metrics.Topic.
-	Topic string `mapstructure:"topic"`
-
-	// Encoding holds the expected encoding of messages (default "otlp_proto")
-	//
-	// Encoding has no default. If explicitly specified, it will take precedence
-	// over the default values of Logs.Encoding, Traces.Encoding, and
-	// Metrics.Encoding.
-	//
-	// Deprecated [v0.124.0]: Use Logs.Encoding, Traces.Encoding, and
-	// Metrics.Encoding.
-	Encoding string `mapstructure:"encoding"`
-
 	// MessageMarking controls the way the messages are marked as consumed.
 	MessageMarking MessageMarking `mapstructure:"message_marking"`
 
@@ -70,41 +52,6 @@ func (c *Config) Unmarshal(conf *confmap.Conf) error {
 	if err := conf.Unmarshal(c); err != nil {
 		return err
 	}
-	// Check if deprecated fields have been explicitly set,
-	// in which case they should be used instead of signal-
-	// specific defaults.
-	var zeroConfig Config
-	if err := conf.Unmarshal(&zeroConfig); err != nil {
-		return err
-	}
-	if c.Topic != "" {
-		if zeroConfig.Logs.Topic == "" {
-			c.Logs.Topic = c.Topic
-		}
-		if zeroConfig.Metrics.Topic == "" {
-			c.Metrics.Topic = c.Topic
-		}
-		if zeroConfig.Traces.Topic == "" {
-			c.Traces.Topic = c.Topic
-		}
-		if zeroConfig.Profiles.Topic == "" {
-			c.Profiles.Topic = c.Topic
-		}
-	}
-	if c.Encoding != "" {
-		if zeroConfig.Logs.Encoding == "" {
-			c.Logs.Encoding = c.Encoding
-		}
-		if zeroConfig.Metrics.Encoding == "" {
-			c.Metrics.Encoding = c.Encoding
-		}
-		if zeroConfig.Traces.Encoding == "" {
-			c.Traces.Encoding = c.Encoding
-		}
-		if zeroConfig.Profiles.Encoding == "" {
-			c.Profiles.Encoding = c.Encoding
-		}
-	}
 
 	// Set OnPermanentError default value to inherit from OnError for backward compatibility
 	// Only if OnPermanentError was not explicitly set in the config

receiver/kafkareceiver/config_test.go

@@ -31,104 +31,6 @@ func TestLoadConfig(t *testing.T) {
 		expected    component.Config
 		expectedErr error
 	}{
-		{
-			id: component.NewIDWithName(metadata.Type, ""),
-			expected: &Config{
-				ClientConfig: func() configkafka.ClientConfig {
-					config := configkafka.NewDefaultClientConfig()
-					config.Brokers = []string{"foo:123", "bar:456"}
-					config.ResolveCanonicalBootstrapServersOnly = true
-					config.ClientID = "the_client_id"
-					return config
-				}(),
-				ConsumerConfig: func() configkafka.ConsumerConfig {
-					config := configkafka.NewDefaultConsumerConfig()
-					config.GroupID = "the_group_id"
-					return config
-				}(),
-				Logs: TopicEncodingConfig{
-					Topic:    "spans",
-					Encoding: "otlp_proto",
-				},
-				Metrics: TopicEncodingConfig{
-					Topic:    "spans",
-					Encoding: "otlp_proto",
-				},
-				Traces: TopicEncodingConfig{
-					Topic:    "spans",
-					Encoding: "otlp_proto",
-				},
-				Profiles: TopicEncodingConfig{
-					Topic:    "spans",
-					Encoding: "otlp_proto",
-				},
-				Topic: "spans",
-				ErrorBackOff: configretry.BackOffConfig{
-					Enabled: false,
-				},
-				Telemetry: TelemetryConfig{
-					Metrics: MetricsConfig{
-						KafkaReceiverRecordsDelay: MetricConfig{
-							Enabled: true,
-						},
-					},
-				},
-			},
-		},
-		{
-			id: component.NewIDWithName(metadata.Type, "legacy_topic"),
-			expected: &Config{
-				ClientConfig:   configkafka.NewDefaultClientConfig(),
-				ConsumerConfig: configkafka.NewDefaultConsumerConfig(),
-				Logs: TopicEncodingConfig{
-					Topic:    "legacy_topic",
-					Encoding: "otlp_proto",
-				},
-				Metrics: TopicEncodingConfig{
-					Topic:    "metrics_topic",
-					Encoding: "otlp_proto",
-				},
-				Traces: TopicEncodingConfig{
-					Topic:    "legacy_topic",
-					Encoding: "otlp_proto",
-				},
-				Profiles: TopicEncodingConfig{
-					Topic:    "legacy_topic",
-					Encoding: "otlp_proto",
-				},
-				Topic: "legacy_topic",
-				ErrorBackOff: configretry.BackOffConfig{
-					Enabled: false,
-				},
-			},
-		},
-		{
-			id: component.NewIDWithName(metadata.Type, "legacy_encoding"),
-			expected: &Config{
-				ClientConfig:   configkafka.NewDefaultClientConfig(),
-				ConsumerConfig: configkafka.NewDefaultConsumerConfig(),
-				Logs: TopicEncodingConfig{
-					Topic:    "otlp_logs",
-					Encoding: "legacy_encoding",
-				},
-				Metrics: TopicEncodingConfig{
-					Topic:    "otlp_metrics",
-					Encoding: "metrics_encoding",
-				},
-				Traces: TopicEncodingConfig{
-					Topic:    "otlp_spans",
-					Encoding: "legacy_encoding",
-				},
-				Profiles: TopicEncodingConfig{
-					Topic:    "otlp_profiles",
-					Encoding: "legacy_encoding",
-				},
-				Encoding: "legacy_encoding",
-				ErrorBackOff: configretry.BackOffConfig{
-					Enabled: false,
-				},
-			},
-		},
 		{
 			id: component.NewIDWithName(metadata.Type, "logs"),
 			expected: &Config{

receiver/kafkareceiver/testdata/config.yaml

@@ -1,23 +1,3 @@
-kafka:
-  topic: spans
-  brokers:
-    - "foo:123"
-    - "bar:456"
-  resolve_canonical_bootstrap_servers_only: true
-  client_id: the_client_id
-  group_id: the_group_id
-  telemetry:
-    metrics:
-      kafka_receiver_records_delay:
-        enabled: true
-kafka/legacy_topic:
-  topic: legacy_topic
-  metrics:
-    topic: metrics_topic
-kafka/legacy_encoding:
-  encoding: legacy_encoding
-  metrics:
-    encoding: metrics_encoding
 kafka/logs:
   logs:
     topic: logs