### Example: Get API Versions

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/client.md

Demonstrates how to fetch API versions from a broker and iterate through the results.

```go
resp, err := client.ApiVersions(context.Background(), &kafka.ApiVersionsRequest{})
if err != nil {
    log.Fatal(err)
}

for _, av := range resp.ApiKeys {
    fmt.Printf("API %d: versions %d-%d\n", av.ApiKey, av.MinVersion, av.MaxVersion)
}
```

--------------------------------

### Release Example Usage

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Shows how to obtain a generation, defer its release, and start processing messages within the generation's context.

```go
gen, err := group.Next(ctx)
if err != nil {
    log.Fatal(err)
}
defer gen.Release(ctx)

err = gen.Start(ctx, func(ctx context.Context) error {
    // Process messages
    return nil
})
```

--------------------------------

### Example: LookupPartitions

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/dialer.md

Example of how to use the LookupPartitions function to retrieve topic partition information.

```go
ctx := context.Background()
partitions, err := kafka.LookupPartitions(ctx, "tcp", "localhost:9092", "my-topic")
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Topic has %d partitions\n", len(partitions))
```

--------------------------------

### WriteMessages Example

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/writer.md

Example demonstrating how to use the WriteMessages method to send multiple messages to Kafka.

```go
ctx := context.Background()
err := w.WriteMessages(ctx,
    kafka.Message{
        Key:   []byte("key1"),
        Value: []byte("message1"),
    },
    kafka.Message{
        Key:   []byte("key2"),
        Value: []byte("message2"),
    },
)
if err != nil {
    log.Fatal(err)
}
```

--------------------------------

### Example Writer Compression Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/types.md

Demonstrates how to configure a Kafka writer to use Gzip compression.

```go
w := &kafka.Writer{
    Compression: compress.Gzip(),
}
```

--------------------------------

### Full Consumer Group Example

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

This example demonstrates the complete lifecycle of a Kafka consumer group, including initialization, message processing across assigned partitions, and offset management. Use this for complex applications requiring explicit control over rebalances and offset commits.

```go
package main

import (
    "context"
    "log"
    "github.com/segmentio/kafka-go"
)

func main() {
    // Create consumer group
    group, err := kafka.NewConsumerGroup(kafka.ConsumerGroupConfig{
        ID:      "my-group",
        Brokers: []string{"localhost:9092"},
        Topics:  []string{"my-topic"},
    })
    if err != nil {
        log.Fatal(err)
    }
    defer group.Close()

    ctx := context.Background()
    
    // Process generations
    for {
        gen, err := group.Next(ctx)
        if err != nil {
            log.Fatal(err)
        }

        // Process this generation
        err = gen.Start(ctx, func(ctx context.Context) error {
            assignments := gen.Assignments()
            
            // Create readers for assigned partitions
            readers := make(map[string]map[int]*kafka.Reader)
            for topic, partitions := range assignments {
                readers[topic] = make(map[int]*kafka.Reader)
                for _, partition := range partitions {
                    r := kafka.NewReader(kafka.ReaderConfig{
                        Brokers:   []string{"localhost:9092"},
                        Topic:     topic,
                        Partition: partition,
                    })
                    readers[topic][partition] = r
                }
            }
            defer closeReaders(readers)

            // Read and process messages
            offsets := make(map[string]map[int]int64)
            for {
                select {
                case <-ctx.Done():
                    // Rebalance occurred
                    return ctx.Err()
                default:
                    // Read from assigned partitions
                    for topic, partitionReaders := range readers {
                        for partition, r := range partitionReaders {
                            msg, err := r.FetchMessage(ctx)
                            if err != nil {
                                continue
                            }
                            
                            // Process message
                            handleMessage(msg)
                            
                            // Track offset
                            if offsets[topic] == nil {
                                offsets[topic] = make(map[int]int64)
                            }
                            offsets[topic][partition] = msg.Offset + 1
                        }
                    }
                    
                    // Commit offsets periodically
                    if err := gen.CommitOffsets(ctx, offsets); err != nil {
                        return err
                    }
                }
            }
        })

        if err != nil {
            log.Printf("Error in generation: %v", err)
        }

        // Release generation for next rebalance
        gen.Release(ctx)
    }
}

func closeReaders(readers map[string]map[int]*kafka.Reader) {
    for _, partitionReaders := range readers {
        for _, r := range partitionReaders {
            r.Close()
        }
    }
}

func handleMessage(msg kafka.Message) {
    // Process message
}
```

--------------------------------

### Create a Kafka Message

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/types.md

Example of creating a Message instance with key, value, and headers.

```Go
msg := kafka.Message{
    Key:   []byte("user-123"),
    Value: []byte(`{"name":"John","age":30}`),
    Headers: []kafka.Header{
        {Key: "content-type", Value: []byte("application/json")},
    },
}
```

--------------------------------

### Example Version Marshal

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/types.md

Demonstrates how to marshal metadata using a specific Kafka API version.

```go
v := kafka.Version(5)
data, err := v.Marshal(metadata)
```

--------------------------------

### Recommended Writer Configuration Example

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/writer.md

Demonstrates direct configuration of the Writer struct, which is the recommended approach for initializing a Kafka writer.

```go
w := &kafka.Writer{
    Addr:     kafka.TCP("localhost:9092", "localhost:9093"),
    Topic:    "my-topic",
    Balancer: &kafka.LeastBytes{},
}
defer w.Close()

err := w.WriteMessages(context.Background(),
    kafka.Message{
        Key:   []byte("key1"),
        Value: []byte("value1"),
    },
)
if err != nil {
    log.Fatal(err)
}
```

--------------------------------

### Kafka-Go Client Example

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/configuration.md

Illustrates creating a Kafka client with a specified address and timeout, and making a metadata request.

```go
client := &kafka.Client{
    Addr:    kafka.TCP("localhost:9092"),
    Timeout: 30*time.Second,
}

resp, err := client.Metadata(ctx, &kafka.MetadataRequest{
    Topics: []string{"my-topic"},
})
```

--------------------------------

### Install AWS MSK IAM V2 module

Source: https://github.com/segmentio/kafka-go/blob/main/sasl/aws_msk_iam_v2/README.md

Add the extension to your project dependencies using the go get command.

```shell
go get github.com/segmentio/kafka-go/sasl/aws_msk_iam_v2
```

--------------------------------

### Assignments Example Usage

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Iterates through the partition assignments returned by the Assignments method and prints them.

```go
assignments := gen.Assignments()
for topic, partitions := range assignments {
    fmt.Printf("Topic %s: partitions %v\n", topic, partitions)
}
```

--------------------------------

### ConsumerGroupConfig Example

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Demonstrates how to create a ConsumerGroupConfig with custom group balancers. Ensure Kafka brokers and topics are correctly specified.

```go
config := kafka.ConsumerGroupConfig{
    ID:         "my-group",
    Brokers:    []string{"localhost:9092"},
    Topics:     []string{"my-topic"},
    GroupBalancers: []kafka.GroupBalancer{
        &kafka.RangeGroupBalancer{},
        &kafka.RoundRobinGroupBalancer{},
    },
}

group, err := kafka.NewConsumerGroup(config)
if err != nil {
    log.Fatal(err)
}
```

--------------------------------

### Run Kafka Locally with Docker

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Starts a local Kafka environment using docker-compose.

```bash
docker-compose up -d
```

--------------------------------

### Example: Reading Messages from a Specific Partition

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/reader.md

Demonstrates how to create a reader for a specific topic and partition, and then continuously read messages. Ensure to close the reader when done.

```go
reader := kafka.NewReader(kafka.ReaderConfig{
    Brokers:   []string{"localhost:9092"},
    Topic:     "my-topic",
    Partition: 0,
    MaxBytes:  10e6, // 10MB
})
defers reader.Close()

for {
    msg, err := reader.ReadMessage(context.Background())
    if err != nil {
        break
    }
    fmt.Printf("Message: %s\n", string(msg.Value))
}
```

--------------------------------

### Using RackAffinityGroupBalancer with Fallback

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/balancers.md

Example of configuring a kafka.Reader with RackAffinityGroupBalancer as the primary strategy and RangeGroupBalancer as a fallback.

```go
reader := kafka.NewReader(kafka.ReaderConfig{
    GroupID: "my-group",
    GroupBalancers: []kafka.GroupBalancer{
        &kafka.RackAffinityGroupBalancer{Rack: "us-east-1a"},
        &kafka.RangeGroupBalancer{}, // Fallback
    },
})
```

--------------------------------

### Example Logger Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/types.md

Shows how to configure a Kafka reader to use a custom logger, such as a standard `*log.Logger` writing to stderr.

```go
reader := kafka.NewReader(kafka.ReaderConfig{
    Logger: log.New(os.Stderr, "[kafka] ", log.LstdFlags),
})
```

--------------------------------

### Example: Consuming Messages as a Consumer Group

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/reader.md

Shows how to configure a reader to join a consumer group for automatic partition assignment and offset management. Errors during message reading will break the loop.

```go
reader := kafka.NewReader(kafka.ReaderConfig{
    Brokers: []string{"localhost:9092"},
    GroupID: "my-group",
    Topic:   "my-topic",
})
defers reader.Close()

for {
    msg, err := reader.ReadMessage(context.Background())
    if err != nil {
        break
    }
    fmt.Printf("Group offset: %d\n", msg.Offset)
}
```

--------------------------------

### Run Kafka docker-compose

Source: https://github.com/segmentio/kafka-go/blob/main/docker_compose_versions/README.md

Command to start the Kafka container using a version-specific docker-compose file.

```bash
# docker-compose -f ./docker_compose_versions/docker-compose-<kafka_version>.yml up -d
```

--------------------------------

### Basic Kafka-Go Dialer Example

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/configuration.md

Illustrates how to create and use a basic Kafka dialer with custom client ID, timeout, and keep-alive settings.

```go
dialer := &kafka.Dialer{
    ClientID: "my-app",
    Timeout:  10*time.Second,
    KeepAlive: 30*time.Second,
}

conn, err := dialer.DialContext(ctx, "tcp", "localhost:9092")
```

--------------------------------

### Generation Start Processing

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Starts processing messages for a consumer group generation with automatic offset management. The provided function is executed within a context that is cancelled upon rebalance.

```go
err := gen.Start(ctx, func(ctx context.Context) error {
    for {
        select {
        case <-ctx.Done():
            return nil // Rebalance occurred
        default:
            // Process messages for assigned partitions
        }
    }
})
```

--------------------------------

### CommitOffsets Example Structure

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Defines the structure for specifying topic, partition, and offset mappings when committing offsets.

```go
offsets := map[string]map[int]int64{
    "topic1": {
        0: 100,  // partition 0, offset 100
        1: 200,  // partition 1, offset 200
    },
    "topic2": {
        0: 50,   // partition 0, offset 50
    },
}
```

--------------------------------

### CommitOffsets Example Usage

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Demonstrates how to call the CommitOffsets method with a defined offsets map and handle potential errors.

```go
err := gen.CommitOffsets(ctx, offsets)
if err != nil {
    log.Fatal(err)
}
```

--------------------------------

### Fetch Metadata Example

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/client.md

Retrieves broker and partition metadata for a specified topic. Ensure the context is properly handled for cancellation and timeouts.

```go
resp, err := client.Metadata(context.Background(), &kafka.MetadataRequest{
    Topics: []string{"my-topic"},
})
if err != nil {
    log.Fatal(err)
}

for _, topic := range resp.Topics {
    fmt.Printf("Topic: %s\n", topic.Name)
    for _, part := range topic.Partitions {
        fmt.Printf("  Partition %d: Leader=%d\n", part.ID, part.Leader.ID)
    }
}
```

--------------------------------

### Configure kafka-go Writer with Different Settings

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/configuration.md

Example of configuring the kafka-go writer with different batching, balancing, and acknowledgment settings.

```go
w := &kafka.Writer{
    Addr:        kafka.TCP("broker1:9092", "broker2:9092"),
    Topic:       "logs",
    Balancer:    &kafka.Hash{},
    BatchSize:   500,
    BatchBytes:  5e6,      // 5 MB
    BatchTimeout: 100*time.Millisecond,
    RequiredAcks: kafka.RequireOne,
    Compression: compress.Gzip(),
}
```

--------------------------------

### Create Topics Example

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/client.md

Creates new topics on the Kafka cluster. Specify the topic name, number of partitions, and replication factor.

```go
resp, err := client.CreateTopics(context.Background(), &kafka.CreateTopicsRequest{
    Topics: []kafka.TopicConfig{
        {
            Topic:             "my-topic",
            NumPartitions:     3,
            ReplicationFactor: 1,
        },
    },
})
```

--------------------------------

### Example GroupMemberAssignments Structure

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/types.md

Illustrates the structure of a GroupMemberAssignments map, showing topic-to-partition assignments for multiple members.

```go
assignments := GroupMemberAssignments{
    "member-1": {
        "topic-1": []int{0, 1},
        "topic-2": []int{0},
    },
    "member-2": {
        "topic-1": []int{2, 3},
        "topic-2": []int{1},
    },
}
```

--------------------------------

### Next Generation Example

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Waits for the next consumer group generation and releases it upon completion. Handles potential errors during generation retrieval.

```go
ctx := context.Background()
gen, err := group.Next(ctx)
if err != nil {
    log.Fatal(err)
}
def gen.Release(ctx)
```

--------------------------------

### Example Kafka Read Deadline Usage

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Demonstrates setting a read deadline before attempting to read a message. This is useful for preventing indefinite waits.

```go
conn.SetReadDeadline(time.Now().Add(10 * time.Second))
msg, err := conn.ReadMessage(1024 * 1024)
if err != nil {
    log.Fatal(err)
}
```

--------------------------------

### Read and Print Partition Information

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/types.md

Example of reading partition information for a topic and printing details about each partition's leader, ISR count, and replica count.

```Go
partitions, err := conn.ReadPartitions("my-topic")
for _, p := range partitions {
    fmt.Printf("Partition %d: Leader=%d, ISR=%d, Replicas=%d\n",
        p.ID, p.Leader.ID, len(p.Isr), len(p.Replicas))
}
```

--------------------------------

### Generation.Start

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Starts processing for the current generation, automatically managing offset commits. The provided function `fn` is called for partition processing, and its context is cancelled upon rebalance.

```APIDOC
## Generation.Start

### Description
Starts processing for this generation with automatic offset management.

### Method Signature
```go
func (gen *Generation) Start(ctx context.Context, fn func(ctx context.Context) error) error
```

### Parameters
- **ctx** (context.Context) - Required - Context for cancellation (returns ErrGenerationEnded on rebalance).
- **fn** (func(ctx context.Context) error) - Required - The processing function.

### Example Usage
```go
err := gen.Start(ctx, func(ctx context.Context) error {
    for {
        select {
        case <-ctx.Done():
            return nil // Rebalance occurred
        default:
            // Process messages for assigned partitions
        }
    }
})
if err != nil {
    log.Fatal(err)
}
```

### Returns
- **error** - An error if processing fails.
```

--------------------------------

### Balancer Interface Implementations

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/types.md

Provides example structs for common Balancer interface implementations, including RoundRobin, LeastBytes, and Hash.

```go
type RoundRobin struct {
    ChunkSize int  // Messages per partition before rotating
}

type LeastBytes struct{}

type Hash struct {
    Hasher hash.Hash32
}
```

--------------------------------

### Read First Available Offset

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Retrieves the earliest available offset for the partition. Useful for starting consumption from the beginning.

```go
func (c *Conn) ReadFirstOffset() (int64, error)
```

--------------------------------

### Configure SASL + TLS Dialer

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/dialer.md

This example shows how to configure the dialer for both SASL authentication and TLS encryption. It requires importing the SCRAM mechanism and providing user credentials.

```go
import (
    "crypto/tls"
    "github.com/segmentio/kafka-go/sasl/scram"
)

dialer := &kafka.Dialer{
    TLS: &tls.Config{
        ServerName: "kafka.example.com",
    },
    SASLMechanism: scram.Mechanism{
        User: "user",
        Pass: "password",
    },
}

conn, err := dialer.DialContext(ctx, "tcp", "kafka.example.com:9093")
```

--------------------------------

### Custom Balancer: Deterministic by Message Index

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/balancers.md

Example of a custom BalancerFunc that deterministically routes messages to partitions based on a counter, ensuring even distribution over time.

```go
// Custom balancer: deterministic based on message index
var counter int
w := &kafka.Writer{
    Balancer: kafka.BalancerFunc(func(msg kafka.Message, partitions ...int) int {
        defer func() { counter++ }()
        return partitions[counter % len(partitions)]
    }),
}
```

--------------------------------

### Initialize Kafka Client

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/client.md

Demonstrates the typical way to initialize a Kafka client by directly setting its address.

```go
client := &kafka.Client{
    Addr: kafka.TCP("localhost:9092"),
}
```

--------------------------------

### Custom Balancer: Always Route to Partition 0

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/balancers.md

Example of a custom BalancerFunc that always routes messages to the first available partition (index 0).

```go
// Custom balancer: always route to partition 0
w := &kafka.Writer{
    Balancer: kafka.BalancerFunc(func(msg kafka.Message, partitions ...int) int {
        return partitions[0]
    }),
}
```

--------------------------------

### Dialer SASLMechanism Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/dialer.md

Example of configuring a SASL mechanism for a kafka.Dialer, such as PLAIN authentication.

```go
import "github.com/segmentio/kafka-go/sasl/plain"

dialer := &kafka.Dialer{
    SASLMechanism: plain.Mechanism{
        User: "username",
        Pass: "password",
    },
}
```

--------------------------------

### Configure a Kafka Reader

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/configuration.md

Sets up a basic Kafka reader with broker addresses, topic, partition, fetch limits, and a custom logger.

```go
config := kafka.ReaderConfig{
    Brokers:   []string{"localhost:9092", "localhost:9093"},
    Topic:     "events",
    Partition: 0,
    MaxBytes:  10e6,        // 10 MB
    MinBytes:  1,
    MaxWait:   500*time.Millisecond,
    Logger:    log.New(os.Stderr, "[reader] ", log.LstdFlags),
}

reader := kafka.NewReader(config)
```

--------------------------------

### Using RoundRobinGroupBalancer in ReaderConfig

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/balancers.md

Example of configuring a kafka.Reader to use the RoundRobinGroupBalancer for consumer group rebalancing.

```go
reader := kafka.NewReader(kafka.ReaderConfig{
    GroupID: "my-group",
    GroupBalancers: []kafka.GroupBalancer{
        &kafka.RoundRobinGroupBalancer{},
    },
})
```

--------------------------------

### Using RangeGroupBalancer in ReaderConfig

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/balancers.md

Example of configuring a kafka.Reader to use the RangeGroupBalancer for consumer group rebalancing.

```go
reader := kafka.NewReader(kafka.ReaderConfig{
    GroupID: "my-group",
    GroupBalancers: []kafka.GroupBalancer{
        &kafka.RangeGroupBalancer{},
    },
})
```

--------------------------------

### CommitUncommittedOffsets

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Commits all uncommitted offsets for this generation. It respects context deadlines and can be called from the Start function.

```APIDOC
## CommitUncommittedOffsets

### Description
Commits all uncommitted offsets for this generation.

### Method
N/A (Method on Go struct)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
err := gen.CommitUncommittedOffsets(ctx)
if err != nil {
    log.Fatal(err)
}
```

### Response
#### Success Response (200)
None (returns error)

#### Response Example
None
```

--------------------------------

### Kafka Topic Creation

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/README.md

Shows how to use the administrative client to create Kafka topics with specified partitions and replication factors. Use this for cluster management tasks.

```go
import "github.com/segmentio/kafka-go"

client := &kafka.Client{
    Addr: kafka.TCP("localhost:9092"),
}

err := client.CreateTopics(context.Background(), &kafka.CreateTopicsRequest{
    Topics: []kafka.TopicConfig{
        {
            Topic:             "my-topic",
            NumPartitions:     3,
            ReplicationFactor: 1,
        },
    },
})
```

--------------------------------

### Get Local Connection Address

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Retrieves the local network address associated with the Kafka connection.

```go
func (c *Conn) LocalAddr() net.Addr
```

--------------------------------

### ReadLag

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/reader.md

Fetches the current lag from the broker. This method can be used to get the consumer lag information.

```APIDOC
## ReadLag

### Description
Fetches the current lag from the broker, indicating how many messages are behind the high watermark.

### Method
func (r *Reader) ReadLag(ctx context.Context) (int64, error)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
ctx := context.Background()
lag, err := reader.ReadLag(ctx)
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Current lag: %d\n", lag)
```

### Response
#### Success Response
- **Lag** (int64) - The current lag in messages.

#### Response Example
(See example in code block above)

### Error Handling
- Returns an error if fetching the lag fails.
```

--------------------------------

### Initialize a Kafka Client

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Create a client with a shared transport for managing connection pools.

```go
mechanism, err := scram.Mechanism(scram.SHA512, "username", "password")
if err != nil {
    panic(err)
}

// Transports are responsible for managing connection pools and other resources,
// it's generally best to create a few of these and share them across your
// application.
sharedTransport := &kafka.Transport{
    SASL: mechanism,
}

client := &kafka.Client{
    Addr:      kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
    Timeout:   10 * time.Second,
    Transport: sharedTransport,
}
```

--------------------------------

### Dialer ClientID Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/dialer.md

Example of setting the ClientID for a kafka.Dialer. This unique identifier is used for client connections.

```go
dialer := &kafka.Dialer{
    ClientID: "my-app-producer-1",
}
```

--------------------------------

### Get All Brokers

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Fetches a list of all brokers currently in the Kafka cluster. This is useful for discovering the cluster topology.

```go
func (c *Conn) Brokers() ([]Broker, error)
```

--------------------------------

### Initialize a Kafka Writer

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Configures a writer with a TCP address, topic, and custom logging functions.

```go
func logf(msg string, a ...interface{}) {
	fmt.Printf(msg, a...)
	fmt.Println()
}

w := &kafka.Writer{
	Addr:        kafka.TCP("localhost:9092"),
	Topic:       "topic",
	Logger:      kafka.LoggerFunc(logf),
	ErrorLogger: kafka.LoggerFunc(logf),
}
```

--------------------------------

### Initialize a Kafka Reader

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Configures a reader with specific brokers, topic, partition, and custom logging functions.

```go
func logf(msg string, a ...interface{}) {
	fmt.Printf(msg, a...)
	fmt.Println()
}

r := kafka.NewReader(kafka.ReaderConfig{
	Brokers:     []string{"localhost:9092", "localhost:9093", "localhost:9094"},
	Topic:       "my-topic1",
	Partition:   0,
	Logger:      kafka.LoggerFunc(logf),
	ErrorLogger: kafka.LoggerFunc(logf),
})
```

--------------------------------

### Dialer Resolver Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/dialer.md

Example of configuring a custom Resolver for a kafka.Dialer. This allows for custom broker address resolution.

```go
dialer := &kafka.Dialer{
    Resolver: &customResolver{},
}
```

--------------------------------

### Basic Kafka Producer

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/README.md

Demonstrates how to create a high-level producer with automatic batching, retries, and partition balancing. Use this for simple message production.

```go
import "github.com/segmentio/kafka-go"

w := &kafka.Writer{
    Addr:  kafka.TCP("localhost:9092"),
    Topic: "my-topic",
}
deferr w.Close()

err := w.WriteMessages(context.Background(),
    kafka.Message{Value: []byte("Hello Kafka")}),
)
```

--------------------------------

### Dialer Timeout Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/dialer.md

Example of setting the Timeout for a kafka.Dialer. This specifies the maximum time to wait for a connection to complete.

```go
dialer := &kafka.Dialer{
    Timeout: 10 * time.Second,
}
```

--------------------------------

### Describe Configurations

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/client.md

Describes broker or topic configurations. Use this to retrieve the current configuration settings for brokers or topics.

```go
func (c *Client) DescribeConfigs(ctx context.Context, req *DescribeConfigsRequest) (*DescribeConfigsResponse, error)
```

--------------------------------

### Dialer DialFunc Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/dialer.md

Example of configuring a custom DialFunc for a kafka.Dialer. This allows for custom network connection logic.

```go
dialer := &kafka.Dialer{
    DialFunc: func(ctx context.Context, network, addr string) (net.Conn, error) {
        // Custom connection logic, e.g., through proxy
        return net.Dialer{}.DialContext(ctx, network, addr)
    },
}
```

--------------------------------

### Dialer TLS Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/dialer.md

Example of configuring TLS for a kafka.Dialer. This involves loading X.509 key pairs for secure connections.

```go
import "crypto/tls"

cert, err := tls.LoadX509KeyPair("client.crt", "client.key")
if err != nil {
    log.Fatal(err)
}

dialer := &kafka.Dialer{
    TLS: &tls.Config{
        Certificates: []tls.Certificate{cert},
    },
}
```

--------------------------------

### Configure a Kafka Reader

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Initialize a reader with specific brokers, group ID, and topic using a custom dialer.

```go
dialer := &kafka.Dialer{
    Timeout:   10 * time.Second,
    DualStack: true,
    TLS:       &tls.Config{...tls config...},
}

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:        []string{"localhost:9092", "localhost:9093", "localhost:9094"},
    GroupID:        "consumer-group-id",
    Topic:          "topic-A",
    Dialer:         dialer,
})
```

--------------------------------

### Dialer KeepAlive Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/dialer.md

Example of setting the KeepAlive duration for a kafka.Dialer. This configures the TCP keep-alive period for active connections.

```go
dialer := &kafka.Dialer{
    KeepAlive: 30 * time.Second,
}
```

--------------------------------

### Execute kafka-go test cases

Source: https://github.com/segmentio/kafka-go/blob/main/docker_compose_versions/README.md

Command to run the test suite with specific environment variables for Kafka versioning.

```bash
# go clean -cache; KAFKA_SKIP_NETTEST=1 KAFKA_VERSION=<a.b.c> go test -race -cover ./...;
```

--------------------------------

### Clone and Build kafka-go

Source: https://github.com/segmentio/kafka-go/blob/main/CONTRIBUTING.md

Clone the kafka-go repository and build the project using Go Modules. It's recommended to clone outside of GOPATH.

```bash
mkdir $HOME/src
cd $HOME/src
git clone https://github.com/segmentio/kafka-go.git
cd kafka-go
go build ./...
```

--------------------------------

### Describe Consumer Groups

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/client.md

Describes consumer groups and their members. Use this to get information about specific consumer groups by their IDs.

```go
func (c *Client) DescribeGroups(ctx context.Context, req *DescribeGroupsRequest) (*DescribeGroupsResponse, error)
```

```go
resp, err := client.DescribeGroups(context.Background(), &kafka.DescribeGroupsRequest{
    GroupIDs: []string{"my-group"},
})
if err != nil {
    log.Fatal(err)
}

for _, group := range resp.Groups {
    fmt.Printf("Group: %s, Members: %d\n", group.ID, len(group.Members))
}
```

--------------------------------

### Get Controller Broker

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Retrieves the controller broker for the Kafka cluster. Use this when you need to interact with the cluster's controller.

```go
func (c *Conn) Controller() (broker Broker, err error)
```

```go
controller, err := conn.Controller()
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Cluster controller: %s:%d\n", controller.Host, controller.Port)
```

--------------------------------

### Get Connected Broker Information

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Retrieves the Broker object associated with the current connection. Provides details about the broker endpoint.

```go
func (c *Conn) Broker() Broker
```

--------------------------------

### Basic Kafka Consumer

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/README.md

Shows how to create a high-level consumer for reading messages from a specific partition. Use this for basic message consumption without group coordination.

```go
import "github.com/segmentio/kafka-go"

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:   []string{"localhost:9092"},
    Topic:     "my-topic",
    Partition: 0,
})
deferr r.Close()

for {
    msg, err := r.ReadMessage(context.Background())
    if err != nil {
        break
    }
    fmt.Printf("Message: %s\n", string(msg.Value))
}
```

--------------------------------

### Get Remote Kafka Broker Address

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Retrieves the remote network address, which is the address of the Kafka broker the connection is established with.

```go
func (c *Conn) RemoteAddr() net.Addr
```

--------------------------------

### Configure a Kafka Reader for Consumer Groups

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Initialize a reader with a GroupID to enable broker-managed offsets and automatic message commits.

```go
// make a new reader that consumes from topic-A
r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:   []string{"localhost:9092", "localhost:9093", "localhost:9094"},
    GroupID:   "consumer-group-id",
    Topic:     "topic-A",
    MaxBytes:  10e6, // 10MB
})

for {
    m, err := r.ReadMessage(context.Background())
    if err != nil {
        break
    }
    fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
}

if err := r.Close(); err != nil {
    log.Fatal("failed to close reader:", err)
}
```

--------------------------------

### Create a Kafka Writer

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Initialize a writer using either the direct struct approach or the deprecated NewWriter constructor.

```go
w := kafka.Writer{
    Addr: kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"), 
    Topic:   "topic-A",
    Balancer: &kafka.Hash{},
    Transport: &kafka.Transport{
        TLS: &tls.Config{},
      },
    }
```

```go
dialer := &kafka.Dialer{
    Timeout:   10 * time.Second,
    DualStack: true,
    TLS:       &tls.Config{...tls config...},
}

w := kafka.NewWriter(kafka.WriterConfig{
	Brokers: []string{"localhost:9092", "localhost:9093", "localhost:9094"},
	Topic:   "topic-A",
	Balancer: &kafka.Hash{},
	Dialer:   dialer,
})
```

--------------------------------

### CommitUncommittedOffsets Go Function Signature

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/consumer-group.md

Commits all uncommitted offsets for the current generation. Respects context deadline and can be called from the Start function.

```go
func (gen *Generation) CommitUncommittedOffsets(ctx context.Context) error
```

--------------------------------

### Get API Versions

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Fetches the API versions supported by the connected Kafka broker. This helps in understanding compatibility and available features.

```go
func (c *Conn) ApiVersions() ([]ApiVersion, error)
```

--------------------------------

### Low-Level Kafka Connection

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/README.md

Demonstrates establishing a direct connection to a Kafka broker for low-level message reading. Use this when fine-grained control over the protocol is needed.

```go
import "github.com/segmentio/kafka-go"

conn, err := kafka.Dial("tcp", "localhost:9092")
if err != nil {
    log.Fatal(err)
}
deferr conn.Close()

batch := conn.ReadBatch(10e3, 1e6) // 10KB min, 1MB max
deferr batch.Close()

// Read messages from batch
```

--------------------------------

### NewConnWith

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Creates a new `Conn` instance with advanced configuration options.

```APIDOC
## NewConnWith

### Description
Creates a new connection with advanced configuration options.

### Method
`func NewConnWith(conn net.Conn, config ConnConfig) *Conn`

### Parameters
#### Path Parameters
- **conn** (net.Conn) - Required - Raw network connection
- **config** (ConnConfig) - Required - Connection configuration object

### Request Example
```go
// Assuming ConnConfig is defined elsewhere
// config := kafka.ConnConfig{ ClientID: "my-client", Topic: "my-topic", Partition: 0 }
// conn := kafka.NewConnWith(netConn, config)
```

### ConnConfig Structure
#### Fields
- **ClientID** (string) - Optional - Unique client identifier
- **Topic** (string) - Required - Topic name
- **Partition** (int) - Required - Partition number
- **Broker** (int) - Optional - Broker ID
- **Rack** (string) - Optional - Rack identifier
- **TransactionalID** (string) - Optional - Transactional ID for idempotent delivery
```

--------------------------------

### Establish a Kafka Connection

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Create a connection using a configured Dialer with TLS support.

```go
dialer := &kafka.Dialer{
    Timeout:   10 * time.Second,
    DualStack: true,
    TLS:       &tls.Config{...tls config...},
}

conn, err := dialer.DialContext(ctx, "tcp", "localhost:9093")
```

--------------------------------

### Configure a Kafka Reader for Consumer Group

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/configuration.md

Sets up a Kafka reader for a consumer group, specifying group ID, topic, session timeout, commit interval, and custom group balancers.

```go
config := kafka.ReaderConfig{
    Brokers:        []string{"localhost:9092"},
    GroupID:        "my-group",
    Topic:          "my-topic",
    SessionTimeout: 20*time.Second,
    CommitInterval: 1*time.Second,
    GroupBalancers: []kafka.GroupBalancer{
        &kafka.RangeGroupBalancer{},
        &kafka.RoundRobinGroupBalancer{},
    },
}

reader := kafka.NewReader(config)
```

--------------------------------

### Writer Configuration and Usage

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/writer.md

Demonstrates how to configure and use the kafka-go Writer for sending messages. It highlights direct configuration of the Writer struct fields, which is the recommended approach.

```APIDOC
## Writer Configuration and Usage

### Description
This section details the configuration options for the `kafka.Writer` and provides an example of its direct usage for sending messages.

### Constructor (Recommended Approach)

Directly configure the `kafka.Writer` struct fields.

**Example:**
```go
w := &kafka.Writer{
    Addr:     kafka.TCP("localhost:9092", "localhost:9093"),
    Topic:    "my-topic",
    Balancer: &kafka.LeastBytes{},
}
defer w.Close()

err := w.WriteMessages(context.Background(),
    kafka.Message{
        Key:   []byte("key1"),
        Value: []byte("value1"),
    },
)
if err != nil {
    log.Fatal(err)
}
```

### Writer Configuration Fields:

| Field | Type | Required | Default | Description |
|---|---|---|---|---|
| Addr | net.Addr | Yes | — | Address of the kafka cluster |
| Topic | string | No | "" | Topic to produce messages to |
| Balancer | Balancer | No | RoundRobin | Message distribution strategy |
| MaxAttempts | int | No | 10 | Retry attempts |
| WriteBackoffMin | time.Duration | No | — | Minimum backoff duration for retries |
| WriteBackoffMax | time.Duration | No | — | Maximum backoff duration for retries |
| BatchSize | int | No | 100 | Messages per batch |
| BatchBytes | int64 | No | 1048576 | Max batch size in bytes |
| BatchTimeout | time.Duration | No | 1s | Batch timeout |
| ReadTimeout | time.Duration | No | 10s | Read operation timeout |
| WriteTimeout | time.Duration | No | 10s | Write operation timeout |
| RequiredAcks | RequiredAcks | No | RequireNone | Acknowledgment requirement |
| Async | bool | No | false | Async writing flag |
| Completion | func([]Message, error) | No | nil | Callback for write completion |
| Compression | Compression | No | None | Compression codec |
| Logger | Logger | No | nil | Internal change logger |
| ErrorLogger | Logger | No | nil | Error logger |
| Transport | RoundTripper | No | DefaultTransport | Custom transport |
| AllowAutoTopicCreation | bool | No | false | Auto-create missing topics |
```

--------------------------------

### Create Kafka Topic (auto.create.topics.enable=false)

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Explicitly create topics when auto.create.topics.enable is false. This involves connecting to the controller and using the CreateTopics API.

```go
// to create topics when auto.create.topics.enable='false'
topic := "my-topic"

conn, err := kafka.Dial("tcp", "localhost:9092")
if err != nil {
    panic(err.Error())
}
defer conn.Close()

controller, err := conn.Controller()
if err != nil {
    panic(err.Error())
}
var controllerConn *kafka.Conn
controllerConn, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
if err != nil {
    panic(err.Error())
}
defer controllerConn.Close()


topicConfigs := []kafka.TopicConfig{
    {
        Topic:             topic,
        NumPartitions:     1,
        ReplicationFactor: 1,
    },
}

err = controllerConn.CreateTopics(topicConfigs...)
if err != nil {
    panic(err.Error())
}
```

--------------------------------

### Priority-Ordered GroupBalancer Selection

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/balancers.md

Demonstrates configuring multiple GroupBalancers in a priority order. The first balancer supported by all members will be chosen.

```go
// Try RackAffinity first, fall back to Range
GroupBalancers: []kafka.GroupBalancer{
    &kafka.RackAffinityGroupBalancer{Rack: "us-east-1a"},
    &kafka.RangeGroupBalancer{},
    &kafka.RoundRobinGroupBalancer{},
}
```

--------------------------------

### Example Git Commit Message

Source: https://github.com/segmentio/kafka-go/blob/main/CONTRIBUTING.md

Follow this format for commit messages, including a concise imperative subject line and an optional body referencing GitHub issues.

```text
Add Code of Conduct and Code Contribution Guidelines

Add a full Code of Conduct and Code Contribution Guidelines document. 
Provide description on how best to retrieve code, fork, checkout, and commit changes.

Fixes #688
```

--------------------------------

### Kafka Consumer Group

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/README.md

Illustrates how to set up a consumer that participates in a consumer group for coordinated message consumption across partitions. Use this for scalable and fault-tolerant consumption.

```go
import "github.com/segmentio/kafka-go"

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers: []string{"localhost:9092"},
    GroupID: "my-group",
    Topic:   "my-topic",
})
deferr r.Close()

for {
    msg, err := r.ReadMessage(context.Background())
    if err != nil {
        break
    }
    fmt.Printf("Message at offset %d: %s\n", msg.Offset, string(msg.Value))
}
```

--------------------------------

### List Available Offsets

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/client.md

Lists available offsets for partitions. This method can be used to find the earliest or latest offsets for a topic partition.

```go
func (c *Client) ListOffsets(ctx context.Context, req *ListOffsetsRequest) (*ListOffsetsResponse, error)
```

--------------------------------

### Configure Writer with Sarama Hash Partitioner

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Use the `kafka.Hash` balancer to achieve the same message partitioning behavior as Sarama's `sarama.NewHashPartitioner`.

```go
w := &kafka.Writer{
	Addr:     kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
	Topic:    "topic-A",
	Balancer: &kafka.Hash{},
}
```

--------------------------------

### Vendor Dependencies

Source: https://github.com/segmentio/kafka-go/blob/main/CONTRIBUTING.md

If using vendoring, run 'go mod vendor' to include the new library bits.

```bash
> go mod vendor
```

--------------------------------

### Seek Kafka Connection Offset

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Moves the connection's offset to a new position within the partition. Specify the offset value and the reference point (start, current, or end).

```go
func (c *Conn) Seek(offset int64, whence int) (int64, error)
```

--------------------------------

### Set Initial Read Offset

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/reader.md

SetOffset allows you to specify the next offset to read from, useful for starting reads from the beginning (kafka.FirstOffset) or end (kafka.LastOffset). This method is not available when GroupID is set and should be called before the first read.

```go
reader.SetOffset(kafka.FirstOffset) // Read from start
msg, err := reader.ReadMessage(context.Background())
```

--------------------------------

### Describe Client Quotas

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/client.md

Describes the quotas configured for clients.

```go
func (c *Client) DescribeClientQuotas(ctx context.Context, req *DescribeClientQuotasRequest) (*DescribeClientQuotasResponse, error)
```

--------------------------------

### Kafka-Go Dialer with TLS Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/configuration.md

Shows how to configure a Kafka dialer for secure communication using TLS, including loading client certificates.

```go
certs, err := tls.LoadX509KeyPair("client.crt", "client.key")
if err != nil {
    log.Fatal(err)
}

dialer := &kafka.Dialer{
    ClientID: "my-app-tls",
    TLS: &tls.Config{
        Certificates: []tls.Certificate{certs},
        ServerName:   "kafka.example.com",
    },
}
```

--------------------------------

### Get Kafka Connection Offset

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Retrieves the current offset position and its reference point within the Kafka partition. The reference point indicates whether the offset is absolute, relative to the current position, or relative to the end.

```go
func (c *Conn) Offset() (offset int64, whence int)
```

--------------------------------

### NewConn

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/conn.md

Creates a new `Conn` instance from a raw network connection, topic, and partition.

```APIDOC
## NewConn

### Description
Creates a new connection from a raw network connection.

### Method
`func NewConn(conn net.Conn, topic string, partition int) *Conn`

### Parameters
#### Path Parameters
- **conn** (net.Conn) - Required - Raw network connection
- **topic** (string) - Required - Topic name
- **partition** (int) - Required - Partition number

### Request Example
```go
netConn, err := net.Dial("tcp", "localhost:9092")
if err != nil {
    log.Fatal(err)
}
conn := kafka.NewConn(netConn, "my-topic", 0)
defer conn.Close()
```
```

--------------------------------

### Retrieve Writer Statistics

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/writer.md

Get current statistics about the writer's operation, including connection attempts, successful writes, total messages and bytes written, and errors encountered. This helps in monitoring the writer's performance.

```go
stats := w.Stats()
fmt.Printf("Messages written: %d\n", stats.Messages)
fmt.Printf("Total bytes: %d\n", stats.Bytes)
fmt.Printf("Errors: %d\n", stats.Errors)
```

--------------------------------

### Read with SASL Authentication

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Initialize a reader configured with a SASL-enabled dialer.

```go
mechanism, err := scram.Mechanism(scram.SHA512, "username", "password")
if err != nil {
    panic(err)
}

dialer := &kafka.Dialer{
    Timeout:       10 * time.Second,
    DualStack:     true,
    SASLMechanism: mechanism,
}

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:        []string{"localhost:9092","localhost:9093", "localhost:9094"},
    GroupID:        "consumer-group-id",
    Topic:          "topic-A",
    Dialer:         dialer,
})
```

--------------------------------

### ConsumerGroup Type

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/INDEX.md

The ConsumerGroup type facilitates advanced coordination for consuming messages within a consumer group. It manages rebalance handling through the Generation type and is configured via ConsumerGroupConfig. Key methods include Next(), Close(), Start(), CommitOffsets(), and Release().

```APIDOC
## ConsumerGroup Type

### Description
Enables advanced coordination for consuming messages as part of a Kafka consumer group, handling rebalances and offset management.

### Key Methods
- **Next**: Retrieves the next message from the consumer group.
- **Close**: Closes the consumer group connection.
- **Start**: Starts the consumer group.
- **CommitOffsets**: Commits the offsets for consumed messages.
- **Release**: Releases the current generation.

### Configuration
Configured using `ConsumerGroupConfig` with 14 options, including group ID, session timeouts, and rebalance strategies.
```

--------------------------------

### Write with SASL Authentication

Source: https://github.com/segmentio/kafka-go/blob/main/README.md

Initialize a writer using a shared transport configured with SASL.

```go
mechanism, err := scram.Mechanism(scram.SHA512, "username", "password")
if err != nil {
    panic(err)
}

// Transports are responsible for managing connection pools and other resources,
// it's generally best to create a few of these and share them across your
// application.
sharedTransport := &kafka.Transport{
    SASL: mechanism,
}

w := kafka.Writer{
	Addr:      kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
	Topic:     "topic-A",
	Balancer:  &kafka.Hash{},
	Transport: sharedTransport,
}
```

--------------------------------

### ReferenceHash Balancer Configuration

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/balancers.md

Configures the ReferenceHash balancer, providing a reference implementation similar to Sarama's ReferenceHashPartitioner for consistent key-based partitioning.

```go
type ReferenceHash struct {
    Hasher hash.Hash32  // Default: FNV-1a
}

```

```go
w := &kafka.Writer{
    Balancer: &kafka.ReferenceHash{},
}
```

--------------------------------

### Hash Balancer Configuration and Usage

Source: https://github.com/segmentio/kafka-go/blob/main/_autodocs/balancers.md

Configures and demonstrates the Hash balancer, which uses FNV-1a hash on message keys for consistent partition assignment. Falls back to RoundRobin for nil keys.

```go
type Hash struct {
    Hasher hash.Hash32  // Default: FNV-1a
}
```

```go
w := &kafka.Writer{
    Addr:     kafka.TCP("localhost:9092"),
    Topic:    "events",
    Balancer: &kafka.Hash{},
}

// Messages with same key go to same partition
w.WriteMessages(ctx,
    kafka.Message{Key: []byte("user-123"), Value: []byte("action-1")},
    kafka.Message{Key: []byte("user-123"), Value: []byte("action-2")},
)

```