kafkaexporter: use configkafka, extract producer (#38665)

#### Description

Use configkafka and extract a function for constructing a
sarama.SyncProducer given configkafka.ClientConfig and
configkafka.ProducerConfig.

As part of the change, `client_id` has been updated to default to
"otel-collector" instead of "sarama" by default. This is a breaking
change for anyone relying on client ID for monitoring or quota
management.

#### Link to tracking issue

Part of #38411

#### Testing

Updated unit tests.

Manually tested against Redpanda:

`$ docker run --publish=9093:9093 --health-cmd "rpk cluster health |
grep 'Healthy:.*true'"
docker.redpanda.com/redpandadata/redpanda:v23.1.11 redpanda start
--kafka-addr=internal://0.0.0.0:9092,external://0.0.0.0:9093 --smp=1
--memory=1G --mode=dev-container`

Ran the collector with `kafkametrics -> kafkaexporter -> Redpanda ->
kafkareceiver -> debugexporter`:

```yaml
receivers:
  kafka:
    brokers: [localhost:9093]
  kafkametrics:
    brokers: [localhost:9093]
    collection_interval: 10s
    scrapers: [topics, consumers]

exporters:
  debug:
    verbosity: detailed
  kafka:
    brokers: [localhost:9093]

service:
  pipelines:
    metrics/kafka:
      receivers: [kafka]
      exporters: [debug]
    metrics:
      receivers: [kafkametrics]
      exporters: [kafka]
```

#### Documentation

Updated README.

Author: axw

.chloggen/kafkaexporter-configkafka.yaml

@@ -0,0 +1,30 @@
+# 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: kafkaexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: change default client_id to "otel-collector"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38411]
+
+# (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: |
+  The exporter now uses the "configkafka" package which consolidates
+  common configuration structures and default values. As a result of
+  this change, we update the default client_id value to "otel-collector".
+
+# 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]

exporter/kafkaexporter/README.md

@@ -17,13 +17,13 @@ Kafka exporter exports logs, metrics, and traces to Kafka. This exporter uses a
 that blocks and does not batch messages, therefore it should be used with batch and queued retry
 processors for higher throughput and resiliency. Message payload encoding is configurable.
 
-The following settings are required:
-- `protocol_version` (no default): Kafka protocol version e.g. `2.0.0`.
+There are no required settings.
 
 The following settings can be optionally configured:
 - `brokers` (default = localhost:9092): The list of kafka brokers.
+- `protocol_version` (default = 2.1.0): Kafka protocol version.
 - `resolve_canonical_bootstrap_servers_only` (default = false): Whether to resolve then reverse-lookup broker IPs during startup.
-- `client_id` (default = "sarama"): The client ID to configure the Sarama Kafka client with. The client ID will be used for all produce requests.
+- `client_id` (default = "otel-collector"): The client ID to configure the Kafka client with. The client ID will be used for all produce requests.
 - `topic` (default = otlp_spans for traces, otlp_metrics for metrics, otlp_logs for logs): The name of the default kafka topic to export to. See [Destination Topic](#destination-topic) below for more details.
 - `topic_from_attribute` (default = ""): Specify the resource attribute whose value should be used as the message's topic. See [Destination Topic](#destination-topic) below for more details. 
 - `encoding` (default = otlp_proto): The encoding of the traces sent to kafka. All available encodings:
@@ -92,8 +92,8 @@ The following settings can be optionally configured:
     - `requests_per_second` is the average number of requests per seconds.
 - `producer`
   - `max_message_bytes` (default = 1000000) the maximum permitted size of a message in bytes
-  - `required_acks` (default = 1) controls when a message is regarded as transmitted.   https://pkg.go.dev/github.com/IBM/[email protected]#RequiredAcks
-  - `compression` (default = 'none') the compression used when producing messages to kafka. The options are: `none`, `gzip`, `snappy`, `lz4`, and `zstd` https://pkg.go.dev/github.com/IBM/[email protected]#CompressionCodec
+  - `required_acks` (default = 1) controls when a message is regarded as transmitted.   https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#acks
+  - `compression` (default = 'none') the compression used when producing messages to kafka. The options are: `none`, `gzip`, `snappy`, `lz4`, and `zstd` https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#compression-type
   - `flush_max_messages` (default = 0) The maximum number of messages the producer will send in a single broker request.
 
 Example configuration:
@@ -103,7 +103,6 @@ exporters:
   kafka:
     brokers:
       - localhost:9092
-    protocol_version: 2.0.0
 ```
 
 ## Destination Topic

exporter/kafkaexporter/config.go

@@ -4,38 +4,22 @@
 package kafkaexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter"
 
 import (
-	"fmt"
-	"time"
-
-	"github.com/IBM/sarama"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka/configkafka"
 )
 
+var _ component.Config = (*Config)(nil)
+
 // Config defines configuration for Kafka exporter.
 type Config struct {
 	TimeoutSettings           exporterhelper.TimeoutConfig `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
 	QueueSettings             exporterhelper.QueueConfig   `mapstructure:"sending_queue"`
 	configretry.BackOffConfig `mapstructure:"retry_on_failure"`
-
-	// The list of kafka brokers (default localhost:9092)
-	Brokers []string `mapstructure:"brokers"`
-
-	// ResolveCanonicalBootstrapServersOnly makes Sarama do a DNS lookup for
-	// each of the provided brokers. It will then do a PTR lookup for each
-	// returned IP, and that set of names becomes the broker list. This can be
-	// required in SASL environments.
-	ResolveCanonicalBootstrapServersOnly bool `mapstructure:"resolve_canonical_bootstrap_servers_only"`
-
-	// Kafka protocol version
-	ProtocolVersion string `mapstructure:"protocol_version"`
-
-	// ClientID to configure the Kafka client with. This can be leveraged by
-	// Kafka to enforce ACLs, throttling quotas, and more.
-	ClientID string `mapstructure:"client_id"`
+	configkafka.ClientConfig  `mapstructure:",squash"`
+	Producer                  configkafka.ProducerConfig `mapstructure:"producer"`
 
 	// The name of the kafka topic to export to (default otlp_spans for traces, otlp_metrics for metrics)
 	Topic string `mapstructure:"topic"`
@@ -54,90 +38,4 @@ type Config struct {
 	PartitionMetricsByResourceAttributes bool `mapstructure:"partition_metrics_by_resource_attributes"`
 
 	PartitionLogsByResourceAttributes bool `mapstructure:"partition_logs_by_resource_attributes"`
-
-	// Metadata is the namespace for metadata management properties used by the
-	// Client, and shared by the Producer/Consumer.
-	Metadata Metadata `mapstructure:"metadata"`
-
-	// Producer is the namespaces for producer properties used only by the Producer
-	Producer Producer `mapstructure:"producer"`
-
-	// Authentication defines used authentication mechanism.
-	Authentication configkafka.AuthenticationConfig `mapstructure:"auth"`
-}
-
-// Metadata defines configuration for retrieving metadata from the broker.
-type Metadata struct {
-	// Whether to maintain a full set of metadata for all topics, or just
-	// the minimal set that has been necessary so far. The full set is simpler
-	// and usually more convenient, but can take up a substantial amount of
-	// memory if you have many topics and partitions. Defaults to true.
-	Full bool `mapstructure:"full"`
-
-	// Retry configuration for metadata.
-	// This configuration is useful to avoid race conditions when broker
-	// is starting at the same time as collector.
-	Retry MetadataRetry `mapstructure:"retry"`
-}
-
-// Producer defines configuration for producer
-type Producer struct {
-	// Maximum message bytes the producer will accept to produce.
-	MaxMessageBytes int `mapstructure:"max_message_bytes"`
-
-	// RequiredAcks Number of acknowledgements required to assume that a message has been sent.
-	// https://pkg.go.dev/github.com/IBM/[email protected]#RequiredAcks
-	// The options are:
-	//   0 -> NoResponse.  doesn't send any response
-	//   1 -> WaitForLocal. waits for only the local commit to succeed before responding ( default )
-	//   -1 -> WaitForAll. waits for all in-sync replicas to commit before responding.
-	RequiredAcks sarama.RequiredAcks `mapstructure:"required_acks"`
-
-	// Compression Codec used to produce messages
-	// https://pkg.go.dev/github.com/IBM/[email protected]#CompressionCodec
-	// The options are: 'none', 'gzip', 'snappy', 'lz4', and 'zstd'
-	Compression string `mapstructure:"compression"`
-
-	// The maximum number of messages the producer will send in a single
-	// broker request. Defaults to 0 for unlimited. Similar to
-	// `queue.buffering.max.messages` in the JVM producer.
-	FlushMaxMessages int `mapstructure:"flush_max_messages"`
-}
-
-// MetadataRetry defines retry configuration for Metadata.
-type MetadataRetry struct {
-	// The total number of times to retry a metadata request when the
-	// cluster is in the middle of a leader election or at startup (default 3).
-	Max int `mapstructure:"max"`
-	// How long to wait for leader election to occur before retrying
-	// (default 250ms). Similar to the JVM's `retry.backoff.ms`.
-	Backoff time.Duration `mapstructure:"backoff"`
-}
-
-var _ component.Config = (*Config)(nil)
-
-// Validate checks if the exporter configuration is valid
-func (cfg *Config) Validate() error {
-	if cfg.Producer.RequiredAcks < -1 || cfg.Producer.RequiredAcks > 1 {
-		return fmt.Errorf("producer.required_acks has to be between -1 and 1. configured value %v", cfg.Producer.RequiredAcks)
-	}
-	_, err := saramaProducerCompressionCodec(cfg.Producer.Compression)
-	return err
-}
-
-func saramaProducerCompressionCodec(compression string) (sarama.CompressionCodec, error) {
-	switch compression {
-	case "none":
-		return sarama.CompressionNone, nil
-	case "gzip":
-		return sarama.CompressionGZIP, nil
-	case "snappy":
-		return sarama.CompressionSnappy, nil
-	case "lz4":
-		return sarama.CompressionLZ4, nil
-	case "zstd":
-		return sarama.CompressionZSTD, nil
-	default:
-		return sarama.CompressionNone, fmt.Errorf("producer.compression should be one of 'none', 'gzip', 'snappy', 'lz4', or 'zstd'. configured value %v", compression)
-	}
 }