### DescribeBroker Operation Example: Show All Configs

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/07-broker-operations.md

Example of how to retrieve and display all configuration entries, including defaults, for a specific broker or cluster defaults. The output format can be specified.

```go
flags := broker.DescribeBrokerFlags{AllConfigs: true, OutputFormat: "yaml"}
err := operation.DescribeBroker("default", flags)
```

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

### Setup Pre-commit Hook

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Install the pre-commit package and initialize pre-commit hooks in your repository to automatically check for linter errors before committing code.

```bash
pip install --user pre-commit
pre-commit install
```

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

### Preview Partition Changes Example

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/09-partition-operations.md

Example showing how to use the ValidateOnly flag to preview partition reassignment plans without applying them to the cluster. This is useful for validation.

```go
flags := partition.AlterPartitionFlags{
    Replicas:     []int32{2, 3},
    ValidateOnly: true,
}
err := operation.AlterPartition("my-topic", 5, flags)
// Prints reassignment plan without applying
```

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

### Install kafkactl with Winget

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Install kafkactl using the Windows Package Manager.

```bash
winget install kafkactl
```

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

### DescribeBroker Operation Example: Show Dynamic Configs

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/07-broker-operations.md

Example of how to retrieve and display only the dynamically configured values for a specific broker. Use the broker's ID and specify flags.

```go
operation := &broker.Operation{}
flags := broker.DescribeBrokerFlags{AllConfigs: false}
err := operation.DescribeBroker("1", flags)
```

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

### Reassign Single Partition Example

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/09-partition-operations.md

Example demonstrating how to reassign replicas for a specific topic partition to a new set of brokers. Ensure broker IDs exist in the cluster.

```go
operation := &partition.Operation{}
flags := partition.AlterPartitionFlags{
    Replicas: []int32{1, 2, 3},  // Brokers 1, 2, 3 will hold replicas
}
err := operation.AlterPartition("my-topic", 0, flags)
```

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

### Complete Kafkactl Configuration Example

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/10-configuration-reference.md

A comprehensive example of a kafkactl configuration file, demonstrating settings for multiple contexts, TLS, SASL, schema registry, producer, and consumer configurations.

```yaml
contexts:
  default:
    brokers:
      - localhost:9092
    clientID: kafkactl-dev
    requestTimeout: 5s
    kafkaVersion: 3.2.0

  production:
    brokers:
      - kafka1.prod.example.com:9092
      -kafka2.prod.example.com:9092
      -kafka3.prod.example.com:9092
    
    tls:
      enabled: true
      ca: /etc/kafka/prod-ca.crt
      cert: /etc/kafka/prod-client.crt
      certKey: /etc/kafka/prod-client.key
    
    sasl:
      enabled: true
      username: kafkactl-service
      mechanism: scram-sha256
      # Password retrieved from keyring or prompted
    
    schemaRegistry:
      url: https://schema-registry.prod.example.com
      requestTimeout: 10s
      username: schema-registry-user
      # Password retrieved from keyring or prompted
      tls:
        enabled: true
        insecure: false
    
    protobuf:
      importPaths:
        - /etc/kafkactl/proto
      protoFiles:
        - messages.proto
    
    producer:
      partitioner: hash
      requiredAcks: all
      maxMessageBytes: 1000000
    
    consumer:
      isolationLevel: ReadCommitted
    
    requestTimeout: 10s
    kafkaVersion: 3.2.0

keyring:
  enabled: true
```

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

### Fish Auto Completion Setup

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Source Fish shell completions for the current session. For persistent setup, redirect output to the completions directory.

```fish
kafkactl completion fish | source
```

```fish
kafkactl completion fish > ~/.config/fish/completions/kafkactl.fish
```

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

### Context Configuration Example

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/10-configuration-reference.md

Illustrates how to define multiple named contexts, each with its own set of broker addresses.

```yaml
contexts:
  default:
    brokers:
      - localhost:9092
  production:
    brokers:
      - broker1.example.com:9092
      - broker2.example.com:9092
```

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

### Bash Auto Completion Setup

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Source kafkactl bash completions for the current session. For persistent setup, redirect output to a bash completion directory.

```bash
source <(kafkactl completion bash)
```

```bash
kafkactl completion bash > /etc/bash_completion.d/kafkactl
```

```bash
kafkactl completion bash > /usr/local/etc/bash_completion.d/kafkactl
```

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

### Install kafkactl from AUR

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Install the kafkactl package from the Arch User Repository using an AUR helper like yay.

```bash
yay -S kafkactl
```

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

### Example Generic Token Provider Script

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

An example bash script demonstrating how to generate an OAuth token and its extensions for the generic token provider. This script should be executable and output valid JSON.

```bash
#!/bin/bash

```

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

### Install kafkactl with Homebrew

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Use this command to upgrade kafkactl if you have Homebrew installed.

```bash
brew upgrade kafkactl
```

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

### Install kafkactl using Homebrew

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Use this command to install the kafkactl CLI tool via the Homebrew package manager.

```bash
# install kafkactl
brew install kafkactl
```

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

### GetBrokers Operation Example

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/07-broker-operations.md

Example of how to list all brokers in a Kafka cluster using the GetBrokers operation. Specify the output format via flags.

```go
operation := &broker.Operation{}
flags := broker.GetBrokersFlags{OutputFormat: "json"}
err := operation.GetBrokers(flags)
```

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

### Example Usage of DescribeUser

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/08-user-operations.md

Demonstrates how to call the DescribeUser operation with custom flags and a username.

```go
operation := &user.Operation{}
flags := user.DescribeUserFlags{OutputFormat: "yaml"}
err := operation.DescribeUser("alice", flags)
```

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

### Zsh Auto Completion Setup

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Enable Zsh completions by sourcing the output to a file in your fpath. Ensure compinit is loaded in your ~/.zshrc.

```zsh
echo "autoload -U compinit; compinit" >> ~/.zshrc
```

```zsh
kafkactl completion zsh > "${fpath[1]}/_kafkactl"
```

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

### Compile kafkactl from source

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Compile and install kafkactl from its source code using the Go toolchain. Ensure `kafkactl` is in your PATH for auto-completion to work.

```bash
go install github.com/deviceinsight/kafkactl/v5@latest
```

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

### Glob Pattern Examples

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Examples of using glob patterns for filtering messages.

```bash
# match keys starting with "user-"
--filter-key "user-*"

# match keys for user or admin
--filter-key "{user,admin}-*"

# match values containing "error" or "warn"
--filter-value "*error*" or --filter-value "*warn*"
```

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

### Grant Topic Read Permission

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Example of granting 'Read' and 'Describe' permissions to a principal on a specific topic.

```go
operation := &acl.Operation{}
flags := acl.CreateACLFlags{
    Principal:   "User:consumer",
    Topic:       "my-topic",
    Operations:  []string{"Read", "Describe"},
    Allow:       true,
}
err := operation.CreateACL(flags, []string{"my-topic"})
```

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

### Example Usage of DescribeConsumerGroup

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/05-consumer-group-operations.md

Demonstrates how to call the DescribeConsumerGroup operation with specific flags to filter by topic and output as JSON.

```go
operation := &consumergroups.ConsumerGroupOperation{}
flags := consumergroups.DescribeConsumerGroupFlags{
    FilterTopic:  "my-topic",
    OutputFormat: "json",
    PrintMembers: true,
}
err := operation.DescribeConsumerGroup(flags, "my-group")
```

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

### Add Second Authentication Mechanism to User

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/08-user-operations.md

Example demonstrating how to add a second authentication mechanism (e.g., SCRAM-SHA-512) to a user who already has one (e.g., SCRAM-SHA-256). The user will then support both mechanisms.

```go
// User "alice" has SCRAM-SHA-256; add SCRAM-SHA-512
flags := user.AlterUserFlags{
    Mechanism: "SCRAM-SHA-512",
    Password:  "same-or-different-password",
}
err := operation.AlterUser("alice", flags)
// Now "alice" has both SCRAM-SHA-256 and SCRAM-SHA-512
```

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

### Kafkactl Command Pattern Example

Source: https://github.com/deviceinsight/kafkactl/blob/main/CLAUDE.md

Illustrates the command pattern where CLI commands in `cmd/` check for Kubernetes enablement and either delegate to a K8s operation or execute a direct operation.

```go
func newGetTopicsCmd() *cobra.Command {
    var flags topic.GetTopicsFlags
    var cmdGetTopics = &cobra.Command{
        Use:   "topics",
        RunE: func(cmd *cobra.Command, args []string) error {
            if internal.IsKubernetesEnabled() {
                return k8s.NewOperation().Run(cmd, args)
            }
            return (&topic.Operation{}).GetTopics(flags)
        },
    }
    // ... flag definitions
}
```

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

### Preview Broker Configuration Changes

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/07-broker-operations.md

Example of previewing configuration changes without applying them by setting ValidateOnly to true. This helps in validating the intended modifications.

```go
flags := broker.AlterBrokerFlags{
    Configs:      []string{"background.threads=12"},
    ValidateOnly: true,
}
err := operation.AlterBroker("2", flags)
```

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

### Set Cluster Default Configurations

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/07-broker-operations.md

Example of setting default configurations for the entire cluster using the 'default' ID. Changes applied here affect all brokers unless overridden.

```go
flags := broker.AlterBrokerFlags{
    Configs: []string{"background.threads=8"},
}
err := operation.AlterBroker("default", flags)
```

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

### List All Topic ACLs

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Example of how to list all ACLs for topics using JSON output format. Ensure the operation is initialized before calling GetACL.

```go
operation := &acl.Operation{}
flags := acl.GetACLFlags{
    Topics:       true,
    OutputFormat: "json",
}
err := operation.GetACL(flags)
```

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

### Produce Messages in JSON Format

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Demonstrates producing messages in JSON format from a file or stdin, with examples for different JSON structures.

```bash
# each line in myfile.json is expected to contain a json object with fields key, value and headers
kafkactl produce my-topic --file=myfile.json --input-format=json
cat myfile.json | kafkactl produce my-topic --input-format=json
echo '{"value": "my-value"}' | kafkactl produce my-topic --input-format=json
echo '{"key": "my-key", "value": "my-value", "headers": {"my-header": "val"}}' | kafkactl produce my-topic --input-format=json
```

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

### Update User Password

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/08-user-operations.md

Example of updating an existing user's password. Ensure the user exists before calling this.

```go
operation := &user.Operation{}
flags := user.AlterUserFlags{
    Password: "new-password",
}
err := operation.AlterUser("alice", flags)
```

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

### Consume Messages from Beginning

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Consume messages starting from the oldest available offset in the topic.

```bash
kafkactl consume my-topic --from-beginning
```

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

### Alter Multiple Broker Configurations

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/07-broker-operations.md

Example of altering multiple configuration properties for a specific broker simultaneously. This is useful for applying several changes at once.

```go
flags := broker.AlterBrokerFlags{
    Configs: []string{
        "background.threads=8",
        "log.cleaner.threads=2",
        "num.replica.fetchers=4",
    },
}
err := operation.AlterBroker("1", flags)
```

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

### Example: Delete All Offsets for Topic

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/05-consumer-group-operations.md

Demonstrates how to delete all committed offsets for a specific topic within a consumer group. Ensure the ConsumerGroupOffsetOperation is initialized.

```go
operation := &consumergroupoffsets.ConsumerGroupOffsetOperation{}
flags := consumergroupoffsets.DeleteConsumerGroupOffsetFlags{
    Topic: "my-topic",
}
err := operation.DeleteConsumerGroupOffset(flags, "my-group")
```

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

### Basic Kafka Consumption

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/03-consumer-operations.md

Initiates basic consumption from a Kafka topic. Set `FromBeginning` to true to start from the earliest message, and `PrintKeys` to true to display message keys. `OutputFormat` can be set to 'json' for structured output.

```go
operation := &consume.Operation{}
flags := consume.Flags{
    FromBeginning: true,
    PrintKeys:     true,
    OutputFormat:  "json",
}
err := operation.Consume("my-topic", flags)
```

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

### List Kafka Topics

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Retrieves a list of all topics in the Kafka cluster. Both 'get topics' and 'list topics' commands achieve this.

```bash
kafkactl get topics
```

```bash
kafkactl list topics
```

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

### Change User Mechanism and Password

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/08-user-operations.md

Example of changing both the authentication mechanism and password for a user. Supported mechanisms include SCRAM-SHA-256 and SCRAM-SHA-512.

```go
flags := user.AlterUserFlags{
    Mechanism: "SCRAM-SHA-512",
    Password:  "new-password",
}
err := operation.AlterUser("alice", flags)
```

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

### Configure Rate Limiting for Producer

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/04-producer-operations.md

Control the production rate by specifying messages per second using RateInSeconds. This example also specifies an input file for messages.

```go
flags := producer.Flags{
    RateInSeconds: 100,  // 100 messages per second
    File:          "messages.txt",
}
```

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

### Grant Consumer Group Read Permission

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Example of granting 'Read' permission to a principal on a specific consumer group.

```go
flags := acl.CreateACLFlags{
    Principal:  "User:consumer",
    Group:      "my-group",
    Operations: []string{"Read"},
    Allow:      true,
}
err := operation.CreateACL(flags, []string{"my-group"})
```

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

### Delete Topic Read Permissions (Preview)

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Example of how to preview the deletion of 'Read' permissions for topic resources. Set ValidateOnly to true for a dry-run.

```go
operation := &acl.Operation{}
flags := acl.DeleteACLFlags{
    Topics:      true,
    Operation:   "Read",
    PatternType: "any",
    ValidateOnly: true,  // Preview first
}
err := operation.DeleteACL(flags)
```

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

### Rebalance Partition Replicas Across Brokers

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/09-partition-operations.md

Use this operation to redistribute partition replicas to balance load across brokers. This example moves a partition's replicas from brokers {1,2,3} to {2,3,4}, setting broker 2 as the new primary.

```Go
operation := &partition.Operation{}

// Move partition 0 from {1,2,3} to {2,3,4}
flags := partition.AlterPartitionFlags{
    Replicas: []int32{2, 3, 4},
}
er := operation.AlterPartition("my-topic", 0, flags)
// Broker 2 gets the primary, 3 and 4 get followers
// Broker 1's replica is removed
```

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

### Grant Topic Prefixed Pattern Permission

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Example of granting 'Read' and 'Write' permissions to a principal on a topic using a prefixed pattern type.

```go
flags := acl.CreateACLFlags{
    Principal:   "User:service",
    Topic:       "events-",
    Operations:  []string{"Read", "Write"},
    Allow:       true,
    PatternType: "prefixed",
}
err := operation.CreateACL(flags, []string{"events-"})
```

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

### Filter Kafka Message Keys

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/03-consumer-operations.md

Use glob patterns to filter Kafka message keys. This example matches keys starting with 'user-'.

```go
FilterKey: "user-*"
```

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

### Create a Kafka Topic using kafkactl Library

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/00-index.md

Demonstrates creating a Kafka topic with specified partitions and replication factor using the kafkactl library. Requires the topic operation and flags to be set up.

```go
import "github.com/deviceinsight/kafkactl/v5/internal/topic"

// Create topic operation
operation := &topic.Operation{}
flags := topic.CreateTopicFlags{
    Partitions:        3,
    ReplicationFactor: 2,
}
err = operation.CreateTopics([]string{"my-topic"}, flags)
```

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

### Filter Kafka Message Keys with Alternatives

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/03-consumer-operations.md

Use glob patterns with alternatives to filter Kafka message keys. This example matches keys starting with either 'user-' or 'admin-'.

```go
FilterKey: "{user,admin}-*"
```

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

### Prefixed Resource Pattern

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Use this pattern to match resources that start with a specific prefix. This is helpful for managing access to a group of related resources, such as topics with a common naming convention.

```go
// Matches "events-v1", "events-v2", "events-transactions", etc.
PatternType: "prefixed"
ResourceName: "events-"
```

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

### Grant Topic Read Permission from Specific Host

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Example of granting 'Read' permission to a principal on a topic, restricted to a specific client host.

```go
flags := acl.CreateACLFlags{
    Principal:   "User:consumer",
    Topic:       "my-topic",
    Hosts:       []string{"192.168.1.100"},
    Operations:  []string{"Read"},
    Allow:       true,
}
err := operation.CreateACL(flags, []string{"my-topic"})
```

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

### Get Kafka Consumer Groups

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/05-consumer-group-operations.md

Lists all consumer groups in the Kafka cluster. Use flags to filter by topic or set the output format.

```go
func (operation *ConsumerGroupOperation) GetConsumerGroups(
    flags GetConsumerGroupFlags,
) error
```

```go
type GetConsumerGroupFlags struct {
    OutputFormat string
    FilterTopic  string
}
```

```go
operation := &consumergroups.ConsumerGroupOperation{}
flags := consumergroups.GetConsumerGroupFlags{
    FilterTopic:  "my-topic",
    OutputFormat: "json",
}
err := operation.GetConsumerGroups(flags)
```

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

### Alter Single Broker Configuration

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/07-broker-operations.md

Example of altering a single configuration property for a specific broker. Ensure the broker ID and configuration key-value pair are correct.

```go
operation := &broker.Operation{}
flags := broker.AlterBrokerFlags{
    Configs: []string{"background.threads=8"},
}
err := operation.AlterBroker("1", flags)
```

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

### Create Kafka Consumer Group with Specific Partition Offsets

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/05-consumer-group-operations.md

Creates a new consumer group and allows specifying exact offsets for particular partitions of a topic. This is useful for precise starting points.

```go
type CreateConsumerGroupFlags struct {
    Topics       []string
    Partitions   []int32
    Offsets      []int64
    OldestOffset bool
    NewestOffset bool
}
```

```go
flags := consumergroups.CreateConsumerGroupFlags{
    Topics:     []string{"my-topic"},
    Partitions: []int32{0, 2},
    Offsets:    []int64{100, 200},
}
err := operation.CreateConsumerGroup(flags, []string{"new-group"})
```

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

### CreateUser with Custom Salt

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/08-user-operations.md

Shows how to provide a custom base64-encoded salt when creating a user.

```go
// Generate your own salt (must be base64-encoded)
flags := user.CreateUserFlags{
    Password: "mypassword",
    Salt:     "YWJjZGVmZ2hpamtsbW5vcA==",  // Custom salt in base64
}
```

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

### Get Brokers

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Retrieves the list of brokers in a Kafka cluster.

```APIDOC
## Getting Brokers

### Description
To get the list of brokers of a kafka cluster use `get brokers`.

### Command
`kafkactl get brokers`

### Examples
```bash
# get the list of brokers
kafkactl get brokers
```
```

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

### Configuration and Helper Functions

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/README.md

Functions for creating configurations and checking Kubernetes enablement.

```APIDOC
## Configuration and Helper Functions

### Description
Includes functions for creating various configurations and checking if Kubernetes is enabled, as well as credential flushing.

### Functions
- **IsKubernetesEnabled()**: Checks if Kubernetes is enabled.
- **FlushCredentials()**: Flushes credentials.
```

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

### Consume Messages from Timestamp (Date Only)

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Consume messages starting from a specific date.

```bash
kafkactl consume my-topic --from-timestamp 2014-04-26
```

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

### Configure Protobuf Files to Load

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/10-configuration-reference.md

List the .proto files that should be loaded by the Protobuf compiler. These are the primary definitions for your messages.

```yaml
protobuf:
  protoFiles:
    - messages.proto
    - events.proto
```

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

### Delete User (SHA-256 Only)

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/08-user-operations.md

Example of deleting only the SCRAM-SHA-256 credentials for a user named 'alice'.

```go
operation := &user.Operation{}
flags := user.DeleteUserFlags{
    Mechanism: "SCRAM-SHA-256",
}
err := operation.DeleteUser("alice", flags)
```

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

### Build and Run kafkactl

Source: https://github.com/deviceinsight/kafkactl/blob/main/CLAUDE.md

Commands for building, formatting, linting, testing, and cleaning the kafkactl project. CGO is disabled, and tools are managed via go.mod.

```bash
# Build the binary
make build
```

```bash
# Run all checks (fmt, lint, cve-check, test, build, docs)
make all
```

```bash
# Format code
make fmt                    # runs gofmt + goimports
```

```bash
# Lint (golangci-lint via Go tool directive)
make lint                   # runs: go tool golangci-lint run
```

```bash
# Unit tests only (skips integration tests)
make test                   # runs: go tool gotestsum --format testname --hide-summary=skipped -- -v -short ./...
```

```bash
# Integration tests (requires Docker)
make integration_test       # starts docker-compose cluster, runs tests, tears down
```

```bash
# CVE check
make cve-check              # runs: go tool govulncheck ./...
```

```bash
# Clean
make clean
```

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

### Deny Topic Write Permission

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Example of denying 'Write' permission to a principal on a specific topic.

```go
flags := acl.CreateACLFlags{
    Principal:   "User:untrusted",
    Topic:       "sensitive-topic",
    Operations:  []string{"Write"},
    Deny:        true,
}
err := operation.CreateACL(flags, []string{"sensitive-topic"})
```

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

### Consume Messages from Timestamp (Epoch Milliseconds)

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Consume messages starting from a specific epoch millisecond timestamp.

```bash
kafkactl consume my-topic --from-timestamp 1384216367189
```

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

### Create Kafka Topic

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/README.md

Use this snippet to create a new Kafka topic with specified partitions, replication factor, and retention configuration.

```go
// From 02-topic-operations.md

operation := &topic.Operation{}
flags := topic.CreateTopicFlags{
    Partitions:        3,
    ReplicationFactor: 2,
    Configs:           []string{"retention.ms=3600000"},
}
err := operation.CreateTopics([]string{"my-topic"}, flags)
```

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

### Configure Protobuf Sets to Load

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/10-configuration-reference.md

Specify compiled .protoset files to load. These files contain pre-compiled Protobuf definitions, which can speed up loading.

```yaml
protobuf:
  protosetFiles:
    - /usr/local/protos/messages.protoset
```

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

### Describe Kafka Broker Configuration

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

View the configuration details for a specific Kafka broker or the cluster defaults. Use `--all-configs` to see static configurations.

```bash
kafkactl describe broker 1
```

```bash
kafkactl describe broker 1 --all-configs
```

```bash
kafkactl describe broker default
```

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

### Get Kafka Brokers

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Retrieve the list of brokers in a Kafka cluster. This command is essential for understanding the cluster topology.

```bash
kafkactl get brokers
```

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

### Consume Messages from Timestamp (RFC 3339)

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Consume messages starting from a specific timestamp in RFC 3339 format.

```bash
kafkactl consume my-topic --from-timestamp 2009-08-12T22:15:09Z
```

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

### Delete All Topic ACLs

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Example for deleting all ACLs associated with topic resources. Uses 'any' for operation and pattern type.

```go
flags := acl.DeleteACLFlags{
    Topics:       true,
    Operation:   "any",
    PatternType: "any",
}
err := operation.DeleteACL(flags)
```

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

### Get Topics in Kafka Cluster

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/02-topic-operations.md

Lists all topics within a Kafka cluster. Specify output format using GetTopicsFlags.

```Go
func (operation *Operation) GetTopics(flags GetTopicsFlags) error
```

```Go
type GetTopicsFlags struct {
    OutputFormat string
}
```

```Go
operation := &topic.Operation{}
flags := topic.GetTopicsFlags{OutputFormat: "json"}
err := operation.GetTopics(flags)
```

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

### Consume Messages within Timestamp Range

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Consume messages between a start and end timestamp. The offsets are computed at the beginning of the process.

```bash
kafkactl consume my-topic --from-timestamp 2017-07-19T03:30:00 --to-timestamp 2017-07-19T04:30:00
```

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

### Running Go Tests

Source: https://github.com/deviceinsight/kafkactl/blob/main/CLAUDE.md

Commands for running unit and integration tests for the kafkactl project. Integration tests require a Docker Kafka cluster.

```bash
# Unit tests only (integration tests are skipped via -short flag)
go test -v -short ./...
```

```bash
# Integration tests (requires Docker Kafka cluster running first)
cd docker && docker compose up -d
go test -v -run Integration ./...
```

```bash
# Stop integration test cluster when done
cd docker && docker compose down
```

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

### Any Resource Pattern

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Use this pattern to match all resources of a specified type. This is typically used for granting broad permissions, such as allowing a user to access all topics.

```go
// Matches all topics
PatternType: "any"
Topics: true
```

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

### Create Sarama Kafka Client

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/11-common-operations.md

Creates a Sarama Kafka client from a pre-configured ClientContext. Ensure the context is valid before calling.

```go
context, _ := internal.CreateClientContext()
client, err := internal.CreateClient(&context)
if err != nil {
    return err
}
defer client.Close()
```

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

### Initialize kafkactl Client Context

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/00-index.md

Initializes the client context for kafkactl operations. Assumes a configuration file is present. Handles potential errors during context creation.

```go
import "github.com/deviceinsight/kafkactl/v5/internal"

// Initialize configuration (assumes config file exists)
ctx, err := internal.CreateClientContext()
if err != nil {
    return err
}
```

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

### Consume Kafka Messages

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/README.md

Consume messages from a Kafka topic, with options to start from the beginning, print keys, and format output as JSON.

```go
// From 03-consumer-operations.md

operation := &consume.Operation{}
flags := consume.Flags{
    FromBeginning: true,
    PrintKeys:     true,
    OutputFormat:  "json",
}
err := operation.Consume("my-topic", flags)
```

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

### Create and Alter Kafka User with SCRAM Mechanisms

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/08-user-operations.md

Demonstrates creating a user with a specific SCRAM mechanism and then adding another SCRAM mechanism to the same user. The user can then authenticate using either mechanism.

```go
// Create user with SHA-256
operation.CreateUser("alice", CreateUserFlags{
    Mechanism: "SCRAM-SHA-256",
    Password:  "password",
})

// Add SHA-512 mechanism to same user
operation.AlterUser("alice", AlterUserFlags{
    Mechanism: "SCRAM-SHA-512",
    Password:  "password",
})

// Now "alice" can authenticate with either mechanism
// Describe shows both mechanisms
```

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

### Consume Messages from Timestamp (ISO 8601 with Timezone)

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Consume messages starting from a specific timestamp in ISO 8601 format with timezone.

```bash
kafkactl consume my-topic --from-timestamp 2017-07-19T03:21:51
```

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

### Create Kafka Topic

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Creates a new Kafka topic with specified partitions, replication factor, and configurations. Configuration can also be cloned from an existing topic.

```bash
kafkactl create topic my-topic
```

```bash
kafkactl create topic my-topic --partitions 32
```

```bash
kafkactl create topic my-topic --replication-factor 3
```

```bash
kafkactl create topic my-topic --config retention.ms=3600000 --config=cleanup.policy=compact
```

```bash
kafkactl describe topic my-topic -o json > my-topic-config.json
kafkactl create topic my-topic-clone --file my-topic-config.json
```

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

### Consume Messages from Timestamp (ISO 8601 with Milliseconds)

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Consume messages starting from a specific timestamp in ISO 8601 format with milliseconds.

```bash
kafkactl consume my-topic --from-timestamp 2014-04-26T17:24:37.123Z
```

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

### Create Caching Schema Registry Client

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/11-common-operations.md

Initializes a schema registry client with in-memory caching. Requires SchemaRegistry.URL to be configured in the ClientContext.

```go
context, _ := internal.CreateClientContext()
registry, err := internal.CreateCachingSchemaRegistry(&context)
if err != nil {
    return err
}

// Schemas are cached in memory
schema, _ := registry.GetLatestSchema("my-topic-value")
```

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

### Delete User's SHA-512 Mechanism

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/08-user-operations.md

Example of deleting only the SCRAM-SHA-512 credentials for a user named 'bob'. If 'bob' only had SHA-512, they will be completely deleted.

```go
flags := user.DeleteUserFlags{
    Mechanism: "SCRAM-SHA-512",
}
err := operation.DeleteUser("bob", flags)
// If bob only had SHA-512, bob is completely deleted
// If bob has both SHA-256 and SHA-512, only SHA-512 is removed
```

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

### Enable TLS

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/10-configuration-reference.md

Enable TLS for broker connection. Set to 'true' to activate.

```yaml
tls:
  enabled: true
```

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

### Delete Consumer Group ACLs

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/06-acl-operations.md

Example for deleting all ACLs related to consumer group resources. Uses 'any' for operation and pattern type.

```go
flags := acl.DeleteACLFlags{
    Groups:      true,
    Operation:   "any",
    PatternType: "any",
}
err := operation.DeleteACL(flags)
```

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

### Default Partitioning Scheme with Key

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Explains the default partitioning scheme where the hash of the key determines the partition, ensuring the same key always goes to the same partition.

```bash
# the following 3 messages will all be inserted to the same partition
kafkactl produce my-topic --key=my-key --value=my-value
kafkactl produce my-topic --key=my-key --value=my-value
kafkactl produce my-topic --key=my-key --value=my-value
```

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

### Produce Messages to a Specific Partition

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Illustrates how to specify a target partition for producing messages.

```bash
kafkactl produce my-topic --key=my-key --value=my-value --partition=2
```

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

### Print Object with Format

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/11-common-operations.md

Renders an object into JSON, YAML, or a raw string. Use 'json' for pretty-printed output, 'json-raw' for compact output, 'yaml' for YAML, and 'none' for silent mode.

```go
func PrintObject(object interface{}, format string) error
```

```go
topics := []Topic{{Name: "topic1"}, {Name: "topic2"}}
err := output.PrintObject(topics, "json")
```

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

### Create Kafka Client Context

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/11-common-operations.md

Creates a ClientContext from viper configuration and environment variables. Ensure context name is valid, brokers are specified, and version is parsable.

```go
ctx, err := internal.CreateClientContext()
if err != nil {
    return err
}
// ctx now contains brokers, TLS, SASL, schema registry, etc.
```

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

### Filter Kafka Message Headers

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/03-consumer-operations.md

Filter Kafka messages based on header patterns. This example matches a 'trace-id' header with a specific pattern.

```go
FilterHeader: map[string]string{"trace-id": "abc-???"}
```

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

### Create Sarama Cluster Admin Client

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/11-common-operations.md

Creates a Sarama ClusterAdmin client from a ClientContext for cluster operations. Use defer admin.Close() to ensure resources are released.

```go
context, _ := internal.CreateClientContext()
admin, err := internal.CreateClusterAdmin(&context)
if err != nil {
    return err
}
defer admin.Close()

// Create topic
admin.CreateTopic("my-topic", &sarama.TopicDetail{
    NumPartitions:     3,
    ReplicationFactor: 2,
}, false)
```

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

### Filter Kafka Message Values

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/03-consumer-operations.md

Use glob patterns to filter Kafka message values. This example matches values containing 'error'.

```go
FilterValue: "*error*"
```

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

### Create consumer group with specific offsets

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Create a new consumer group and set initial offsets for its partitions. Offsets can be set to the oldest or newest available, or to a specific value for a given partition. Supports multiple topics.

```bash
# create group with offset for all partitions set to oldest
kafkactl create consumer-group my-group --topic my-topic --oldest
# create group with offset for all partitions set to newest
kafkactl create consumer-group my-group --topic my-topic --newest
# create group with offset for a single partition set to specific offset
kafkactl create consumer-group my-group --topic my-topic --partition 5 --offset 100
# create group for multiple topics with offset for all partitions set to oldest
kafkactl create consumer-group my-group --topic my-topic-a --topic my-topic-b --oldest
```

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

### Store Credentials in Config File

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Example of storing SASL username/password and TLS passphrase directly in the configuration file. Not recommended for production environments.

```yaml
contexts:
  my-cluster:
    sasl:
      enabled: true
      username: my-user
      password: my-password   # stored in plaintext
    tls:
      certKey: my-key.pem
      certKeyPassphrase: my-passphrase   # stored in plaintext
```

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

### Describe Kafka Topic

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Provides a detailed description of a specific Kafka topic, including its configuration and partition details. Use --all-configs to see default configurations and --skip-empty to filter out partitions without messages.

```bash
kafkactl describe topic my-topic
```

```bash
kafkactl describe topic my-topic --all-configs
```

```bash
kafkactl describe topic my-topic --skip-empty
```

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

### Create User with Custom Iterations

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/08-user-operations.md

Creates a new SCRAM user with a custom PBKDF2 iteration count. Higher values increase security but add CPU cost.

```go
flags := user.CreateUserFlags{
    Mechanism:  "SCRAM-SHA-256",
    Password:   "my-secure-password",
    Iterations: 8192,
}
er := operation.CreateUser("charlie", flags)
```

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

### Configure OAuth Token Provider

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/10-configuration-reference.md

Configure the OAuth token provider when using the 'oauth' SASL mechanism. This example shows the Azure AD provider.

```yaml
sasl:
  mechanism: oauth
  tokenProvider:
    plugin: azure
    options:
      tenant: mytenant
```

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

### Create Sarama Client Configuration

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/11-common-operations.md

Generates a Sarama configuration object from a ClientContext, applying TLS, SASL, and Kafka version settings. Caller must set producer return flags if needed.

```go
context, _ := internal.CreateClientContext()
config, err := internal.CreateClientConfig(&context)
if err != nil {
    return err
}

producer, _ := sarama.NewSyncProducer(context.Brokers, config)
defer producer.Close()
```

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

### Get Kafka Services in Kubernetes

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Use kubectl to list services in your Kubernetes cluster and filter for Kafka-related services to identify broker service names.

```bash
kubectl get svc | grep kafka
```

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

### Create and Use TableWriter

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/11-common-operations.md

Initializes a TableWriter for creating aligned tabular output. Remember to call Flush() after writing all rows to display the table.

```go
tw := output.NewTableWriter()
tw.WriteHeader("Name", "Partitions", "Replicas")
for _, topic := range topics {
    tw.Write(topic.Name, topic.Partitions, topic.Replicas)
}
tw.Flush()
```

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

### Consume Messages from Timestamp (ISO 8601 with Hour and Minute)

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Consume messages starting from a specific timestamp in ISO 8601 format with hour and minute.

```bash
kafkactl consume my-topic --from-timestamp 2013-04-01T22:43
```

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

### Produce Protobuf Message with Protoset File

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Produces a message to a Kafka topic using Protobuf serialization with a pre-compiled protoset file. This is an alternative to using proto files directly.

```bash
kafkactl produce <topic> --key '{"fvalue":1.2}' --key-proto-type TopicKey --value '{"producedAt":"2021-12-01T14:10:12Z","num":"1"}' --value-proto-type TopicValue --protoset-file kafkamsg.protoset
```

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

### Consume Messages from Timestamp (ISO 8601 without Milliseconds)

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Consume messages starting from a specific timestamp in ISO 8601 format without milliseconds.

```bash
kafkactl consume my-topic --from-timestamp 2014-04-26T17:24:37.123
```

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

### Configure Generic Token Provider Options

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/10-configuration-reference.md

Configure the generic token provider with a script and arguments. This allows for custom token retrieval logic.

```yaml
tokenProvider:
  plugin: generic
  options:
    script: /usr/local/bin/get-token.sh
    args:
      - kafka-scope
```

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

### Client Creation Functions

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/README.md

Functions for creating and configuring Kafka clients and related contexts.

```APIDOC
## Client Creation Functions

### Description
Provides functions to initialize and configure Kafka clients, including context, configuration, and cluster administration capabilities.

### Functions
- **CreateClientContext()**: Creates a client context.
- **CreateClient()**: Creates a Kafka client instance.
- **CreateClusterAdmin()**: Creates a cluster administrator client.
- **CreateClientConfig()**: Creates a client configuration object.
```

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

### Random Partitioner Configuration

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/04-producer-operations.md

Configure the producer to use the random partitioner for distributing messages randomly across available partitions.

```go
flags := producer.Flags{
    Partitioner: "random",
}
```

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

### Change Preferred Leader for a Partition

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/09-partition-operations.md

Modify the preferred leader for a partition by reordering the replica list. This example changes the leader from broker 1 to broker 3.

```Go
// Current: {1,2,3} with broker 1 as leader
// Change to: {3,1,2} with broker 3 as preferred leader
flags := partition.AlterPartitionFlags{
    Replicas: []int32{3, 1, 2},
}
er := operation.AlterPartition("my-topic", 0, flags)
```

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

### Create Kafka Topic

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/02-topic-operations.md

Use this function to create one or more topics with specified partition count, replication factor, and configurations. Ensure the CreateTopicFlags struct is populated with desired settings.

```go
func (operation *Operation) CreateTopics(
    topics []string,
    flags CreateTopicFlags,
) error
```

```go
type CreateTopicFlags struct {
    Partitions        int32
    ReplicationFactor int16
    ValidateOnly      bool
    File              string
    Configs           []string
}
```

```go
operation := &topic.Operation{}
flags := topic.CreateTopicFlags{
    Partitions:        3,
    ReplicationFactor: 2,
    Configs:           []string{"retention.ms=3600000", "compression.type=snappy"},
}
err := operation.CreateTopics([]string{"my-topic"}, flags)
```

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

### Example: Delete Offset for Single Partition

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/05-consumer-group-operations.md

Shows how to delete the committed offset for a specific partition of a topic within a consumer group. The partition number must be provided.

```go
flags := consumergroupoffsets.DeleteConsumerGroupOffsetFlags{
    Topic:     "my-topic",
    Partition: 1,
}
err := operation.DeleteConsumerGroupOffset(flags, "my-group")
```

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

### Enable Keyring Integration

Source: https://github.com/deviceinsight/kafkactl/blob/main/README.adoc

Configure kafkactl to use OS keyring for storing credentials. Defaults to enabled.

```yaml
keyring:
  enabled: true
```

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

### Get Current Kafka Context

Source: https://github.com/deviceinsight/kafkactl/blob/main/_autodocs/11-common-operations.md

Retrieves the name of the currently active Kafka context. This function checks multiple sources in a specific order to determine the active context.

```go
func GetCurrentContext() (string, error) {
    // ... implementation details ...
    return "default", nil
}
```

```go
contextName, err := global.GetCurrentContext()
if err != nil {
    return err
}
```