# Admin API

Karafka provides application administrative functions via the built-in `Karafka::Admin` module, including topic and consumer group management, cluster administration, and more.

!!! tip "Asynchronous Operation Propagation"

    Many Kafka administrative operations (ACLs, configs, topics) are asynchronous in nature. When an API call returns successfully, this means the controller has accepted the request, not that the change has been fully propagated across the cluster. Configuration changes, ACL updates, and topic modifications may take several seconds to be applied on all brokers, depending on cluster size and network conditions. Always allow time for propagation and verify changes are applied across your cluster before proceeding with dependent operations.

!!! tip "Multi-Cluster Support"

    Karafka Admin supports operations on multiple Kafka clusters. You can either use the default cluster via class methods, or create an Admin instance with custom kafka configuration to target a different cluster. For more details, visit the [Multi-Cluster Setup](#multi-cluster-setup) section.

## Admin API / Configuration

`Karafka::Admin` operates using the default cluster configuration, employing a distinct consumer group name, specifically `karafka_admin`. The Web UI also uses this same consumer group as it uses the Admin API internally. If you are implementing granular Kafka ACLs (Access Control List) permissions, ensure that the `karafka_admin` consumer group is granted the necessary permissions to function effectively. If you are using `karafka-web`, you will also need the same permissions applied to the `karafka_web` group as well.

`Karafka::Admin` gets a consistent prefix alongside all other consumer groups, allowing you to streamline permissions across all the consumer groups associated with that application.

If you wish to reconfigure `Karafka::Admin`, you may alter the `config.admin` scope during the Karafka framework configuration. You may, for example, use an alternative `group_id` to replace the default `karafka_admin` part and alter the `kafka` scope settings.

```ruby
class KarafkaApp < Karafka::App
  setup do |config|
    config.client_id = "my_application-#{Process.pid}-#{Socket.gethostname}"

    config.kafka = {
      'bootstrap.servers': '127.0.0.1:9092'
    }

    # Fetch 20MB at most with admin, instead of the default 5MB
    config.admin.kafka[:'fetch.message.max.bytes'] = 20 * 1_048_576
    # Replace karafka_admin group name component with app_admin
    config.admin.group_id = 'app_admin'
  end
end
```

We **strongly advise against** modifying the `config.admin.kafka` settings in their entirety, as they have been configured to meet the demands of admin operations. For any adjustments, we suggest adopting the granular approach outlined above.

In the case of `kafka` settings, if a given setting is not found, the root `kafka` scope value will be used.

### Admin API / Configuration / Multi-Cluster Setup

Karafka Admin supports two approaches for working with multiple Kafka clusters:

#### Admin API / Configuration / Multi-Cluster Setup / Instance-Based API (Recommended)

Create an Admin instance with custom kafka configuration to perform operations on a different cluster:

```ruby
# Default cluster (backward compatible - uses class methods)
Karafka::Admin.cluster_info
Karafka::Admin.create_topic('my-topic', 1, 1)

# Custom cluster via Admin instance
admin = Karafka::Admin.new(kafka: { 'bootstrap.servers': 'other-cluster:9092' })
admin.cluster_info
admin.create_topic('my-topic', 1, 1)
```

All Admin operations are available on both the class (for the default cluster) and on instances (for custom clusters):

```ruby
# Working with a secondary cluster
secondary_admin = Karafka::Admin.new(
  kafka: { 'bootstrap.servers': 'secondary-cluster:9092' }
)

# Topic operations
secondary_admin.create_topic('events', 6, 3)
secondary_admin.delete_topic('old-events')
secondary_admin.topic_info('events')

# Consumer group operations
secondary_admin.read_lags_with_offsets
secondary_admin.seek_consumer_group('my-group', { 'events' => 0 })
secondary_admin.delete_consumer_group('unused-group')

# Read messages from the secondary cluster
messages = secondary_admin.read_topic('events', 0, 10)
```

!!! note "Configuration Scope"

    The `kafka` configuration and, since 2.6.0, an `external_client` are the two customizable arguments per Admin instance. Global admin settings like `max_wait_time` and `poll_timeout` are shared from `Karafka::App.config.admin`.

`Karafka::Admin.new` also accepts an `external_client:` keyword argument, letting admin operations run on an already-open client (a raw rdkafka admin instance, one wrapped with `Karafka::Connection::Proxy`, or a `Karafka::Connection::Client` of a running consumer) instead of opening a new short-lived connection. The external client's lifecycle stays fully with its owner - it is never configured, started, or closed by the Admin instance. This is a low-level API: the caller is responsible for providing a client capable of the operations invoked.

```ruby
# Read lags of a running consumer via its own client connection
admin = Karafka::Admin.new(external_client: client)
admin.read_lags_with_offsets({ 'my-group' => ['events'] })
```

#### Admin API / Configuration / Multi-Cluster Setup / Environment Variable Approach

Alternatively, you can manage multiple Kafka clusters using the `KARAFKA_BOOT_FILE` environment variable. This variable can point to different Karafka boot files configured for a specific cluster. By changing the value of `KARAFKA_BOOT_FILE`, you can specify which cluster to use when performing any Karafka admin and declarative topics operations.

```ruby
# cluster1.rb

class KarafkaApp < Karafka::App
  setup do |config|
    config.client_id = "my_application-#{Process.pid}-#{Socket.gethostname}"

    config.kafka = {
      'bootstrap.servers': '192.168.1.2:9092'
    }
  end
end
```

```ruby
# cluster2.rb

class KarafkaApp < Karafka::App
  setup do |config|
    config.client_id = "my_application-#{Process.pid}-#{Socket.gethostname}"

    config.kafka = {
      'bootstrap.servers': '192.168.1.10:9092'
    }
  end
end
```

```shell
# Access Karafka console to operate in cluster 1
export KARAFKA_BOOT_FILE=cluster1.rb
bundle exec karafka console

# Same applies to declarative topics management
export KARAFKA_BOOT_FILE=cluster2.rb
bundle exec karafka topics migrate
```

## Admin API / Creating a Topic

```ruby
topic_name = 'my_cool_topic'
partitions_count = 2
replication_factor = 1 # 1 for dev, for prod you want more

Karafka::Admin.create_topic(topic_name, partitions_count, replication_factor)
```

## Admin API / Deleting a Topic

```ruby
topic_name = 'my_cool_topic'

Karafka::Admin.delete_topic(topic_name)
```

## Admin API / Altering a Topic

Karafka provides two distinct approaches for managing and altering topic configurations:

- [Declarative Topics](https://karafka.io/docs/Infrastructure-Declarative-Topics.md) (Recommended for most cases)
- [Admin Configs API](https://karafka.io/docs/Infrastructure-Admin-Configs-API.md) (For lower-level control)

### Admin API / Altering a Topic / Declarative Topics Approach

The Declarative Topics feature provides a high-level, code-based way to manage topic configurations. This approach is recommended for most users as it offers:

```ruby
class KarafkaApp < Karafka::App
  setup do |config|
    # ... other config ...
  end

  routes.draw do
    topic :orders do
      config(
        partitions: 6,
        replication_factor: 3,
        'cleanup.policy': 'compact',
        'retention.ms': 604_800_000 # 7 days
      )
    end
  end
end
```

Apply changes using:

```shell
bundle exec karafka topics migrate
```

### Admin API / Altering a Topic / Admin Configs API Approach

For cases requiring more granular control, you can use the Admin Configs API to directly manage topic configurations:

```ruby
# Describe current topic configuration
resource = Karafka::Admin::Configs::Resource.new(type: :topic, name: 'orders')
topics = Karafka::Admin::Configs.describe(resource)

# Alter topic configuration
resource = Karafka::Admin::Configs::Resource.new(type: :topic, name: 'orders')
resource.set('retention.ms', '7200000')  # Set retention to 2 hours
Karafka::Admin::Configs.alter(resource)
```

## Admin API / Describing Topics

`Karafka::Admin.describe_topics` returns the topic id of each requested topic and its partitions with their leaders, replicas, and in-sync replicas. It describes all the topics in one request and returns an array in the order of the topic names you provide:

```ruby
Karafka::Admin.describe_topics(['events']).first

# {
#   name: 'events',
#   topic_id: '7zVJlh/2SuKCqPUdnbl9nQ',
#   internal: false,
#   partitions: [
#     {
#       partition: 0,
#       leader: { id: 1, host: '127.0.0.1', port: 9092, rack: 'rack-1' },
#       replicas: [{ id: 1, host: '127.0.0.1', port: 9092, rack: 'rack-1' }],
#       isr: [{ id: 1, host: '127.0.0.1', port: 9092, rack: 'rack-1' }]
#     }
#   ],
#   authorized_operations: nil,
#   error: nil
# }
```

A topic that cannot be described, for example because it does not exist, does not fail the whole call. Its entry carries the error in `:error`, has no topic id, and has no partitions. Use it to find the topics that do not exist:

```ruby
Karafka::Admin.describe_topics(%w[events orders])
  .select { |topic| topic[:error] }
  .map { |topic| topic[:name] }
```

`Karafka::Admin::Topics.describe` accepts one or more topic names as arguments and always returns an array, also when you describe one topic. On an Admin instance for a custom cluster, use `#describe_topics`.

Each topic description contains the following keys:

<table>
  <thead>
    <tr>
      <th>Key</th>
      <th>Description</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td><code>:name</code></td>
      <td>The topic name.</td>
    </tr>
    <tr>
      <td><code>:topic_id</code></td>
      <td>The topic id (base64), or <code>nil</code> when the topic could not be described.</td>
    </tr>
    <tr>
      <td><code>:internal</code></td>
      <td>Whether it is an internal topic, for example <code>__consumer_offsets</code>.</td>
    </tr>
    <tr>
      <td><code>:partitions</code></td>
      <td>The topic partitions. Each has <code>:partition</code>, <code>:leader</code>, <code>:replicas</code>, and <code>:isr</code>, with brokers as <code>{ id:, host:, port:, rack: }</code>.</td>
    </tr>
    <tr>
      <td><code>:authorized_operations</code></td>
      <td>The operations this client is authorized to perform on the topic, for example <code>:read</code> or <code>:write</code>. <code>nil</code> unless you pass <code>include_authorized_operations: true</code>.</td>
    </tr>
    <tr>
      <td><code>:error</code></td>
      <td>The <code>Rdkafka::RdkafkaError</code> that explains why the topic could not be described, for example <code>unknown_topic_or_part</code>. <code>nil</code> on success.</td>
    </tr>
  </tbody>
</table>

## Admin API / Getting Cluster Info

```ruby
# Get cluster info and list all the topics
info = Karafka::Admin.cluster_info

puts info.topics.map { |topic| topic[:topic_name] }.join(', ')
```

### Admin API / Getting Cluster Info / Describing the Cluster

`Karafka::Admin.describe_cluster` returns the cluster id, the current controller, and the brokers with their racks. Unlike `Karafka::Admin.cluster_info`, it does not fetch any topic metadata.

```ruby
Karafka::Admin.describe_cluster

# {
#   cluster_id: 'kafka-docker-cluster-1',
#   controller: { id: 1, host: '127.0.0.1', port: 9092, rack: 'rack-1' },
#   brokers: [{ id: 1, host: '127.0.0.1', port: 9092, rack: 'rack-1' }],
#   authorized_operations: nil
# }
```

To also get the operations this client is authorized to perform on the cluster, pass `include_authorized_operations: true`. `:authorized_operations` is then an array of operation names, for example `:describe` or `:alter_configs`:

```ruby
Karafka::Admin.describe_cluster(include_authorized_operations: true)[:authorized_operations]
```

On an Admin instance for a custom cluster, use `#describe_cluster`.

## Admin API / Reading Topic Messages

By using the `.read_topic` method, you can read data from a given topic partition without subscribing to it.

!!! note "Topic Details Require Defined Routes"

    While the returned messages are `Karafka::Messages::Message` objects, they may not hold the correct notion of the topic details unless the given topic is defined in Karafka routes. For topics that are not defined, defaults will be used.

!!! note "`read_topic` Skips Compacted and Transaction Offsets"

    When using the `#read_topic` method in the Karafka Admin API to retrieve messages from a topic partition, this method skips offsets of compacted messages and transactions-related messages. This means that these specific messages will not be fetched or displayed even if they exist in the topic partition. However, while these messages are skipped during retrieval, they are still included in the total counts for those in that partition.

    This behavior implies that if you request a certain number of the most recent messages, say the last 10, and all these 10 messages were either related to transactions or were compacted, then the `#read_topic` method will return no data. Thus, you might find situations where you expect data based on the total count but get none due to this offset-skipping behavior.

### Admin API / Reading Topic Messages / Getting Last N Messages

```ruby
topic = 'my_topic'
partition = 0
how_many = 10

messages = Karafka::Admin.read_topic(topic, partition, how_many)

messages.each do |message|
  puts message.raw_payload
end
```

### Admin API / Reading Topic Messages / Getting Messages From a Given Offset

```ruby
topic = 'my_topic'
partition = 0
how_many = 10
first_offset = 50

messages = Karafka::Admin.read_topic(topic, partition, how_many, first_offset)

messages.each do |message|
  puts message.raw_payload
end
```

### Admin API / Reading Topic Messages / Getting Messages From a Given Time

```ruby
topic = 'my_topic'
partition = 0
how_many = 10
# Read at most 10 messages starting 60 seconds ago
start_at = Time.now - 60

messages = Karafka::Admin.read_topic(topic, partition, how_many, start_at)

messages.each do |message|
  puts message.raw_payload
end
```

## Admin API / Adding Partitions To a Topic

If you want to add partitions to an existing topic, you can use the `create_partitions` admin method:

```ruby
topic_name = 'my-busy-topic'
# Indicates how many partitions we want to end up with
total_partitions = 20

Karafka::Admin.create_partitions(topic_name, total_partitions)
```

This method will create all the additional partitions to reach your desired count.

## Admin API / Reading the Watermark Offsets

Watermark offsets represent the current high and low offsets for each partition in a topic.

- The high-watermark offset is the offset of the last message that has been fully replicated across all of the brokers in the Kafka cluster.
- The low watermark offset is important because it determines the starting point for a new consumer reading from a partition. When a consumer subscribes to a Kafka topic, it must know where to start reading messages from in each partition. A new consumer will start reading from the low watermark offset by default.

Watermark offsets are used to track consumers' progress in a Kafka topic. As a consumer reads messages from a topic, it keeps track of the offset of the last message consumed for each partition. By comparing this offset to the watermark offset, the consumer can determine how far behind it is in processing messages.

You can access this information using the `#read_watermark_offsets` admin method as follows:

```ruby
topic = 'my_topic'
partition = 1

low, high = Karafka::Admin.read_watermark_offsets(topic, partition)

puts "Low watermark offset: #{low}"
puts "High watermark offset: #{high}"
```

## Admin API / Reading Partition Offsets

The `#read_partition_offsets` method is a lower-level, more flexible alternative to `#read_watermark_offsets` that gives you precise control over which offset spec to query for each partition, and optionally applies transactional isolation semantics.

### Admin API / Reading Partition Offsets / Isolation Levels

`Karafka::Admin::IsolationLevels` is a public module holding the two isolation level constants accepted by the `isolation_level:` keyword used throughout this page:

- `Karafka::Admin::IsolationLevels::READ_UNCOMMITTED` - the default; may include messages written by producers that have not yet committed (or have aborted) their transaction
- `Karafka::Admin::IsolationLevels::READ_COMMITTED` - excludes those uncommitted/aborted messages

For non-transactional topics, both values give the same result.

<table>
  <thead>
    <tr>
      <th>Feature</th>
      <th><code>read_watermark_offsets</code></th>
      <th><code>read_partition_offsets</code></th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td>Transactional correctness (LSO)</td>
      <td>No - always returns HWM</td>
      <td>Yes - pass <code>isolation_level</code> for LSO</td>
    </tr>
    <tr>
      <td><code>:max_timestamp</code> spec</td>
      <td>No</td>
      <td>Yes</td>
    </tr>
    <tr>
      <td>Time-based offset lookup</td>
      <td>No</td>
      <td>Yes (ms timestamp as offset)</td>
    </tr>
    <tr>
      <td><code>leader_epoch</code> in result</td>
      <td>No</td>
      <td>Yes</td>
    </tr>
    <tr>
      <td>Multi-partition, multi-topic in one call</td>
      <td>Partial</td>
      <td>Yes</td>
    </tr>
  </tbody>
</table>

For non-transactional topics, `:latest` here and the `read_watermark_offsets` high-watermark return the same value.

Pass a hash where keys are topic names and values are arrays of partition specs. Each spec must include `:partition` and `:offset`. The `:offset` can be `:earliest`, `:latest`, `:max_timestamp`, or an integer Unix timestamp **in milliseconds** (returns the first offset at or after that timestamp).

```ruby
results = Karafka::Admin.read_partition_offsets(
  {
    'events' => [
      { partition: 0, offset: :earliest },
      { partition: 1, offset: :latest }
    ]
  }
)

results.each do |r|
  puts "#{r[:topic]}##{r[:partition]}: offset=#{r[:offset]}, leader_epoch=#{r[:leader_epoch]}"
end

# events#0: offset=0, leader_epoch=0
# events#1: offset=100, leader_epoch=0
```

Each result hash contains:

<table>
  <thead>
    <tr>
      <th>Key</th>
      <th>Type</th>
      <th>Description</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td><code>:topic</code></td>
      <td>String</td>
      <td>Topic name</td>
    </tr>
    <tr>
      <td><code>:partition</code></td>
      <td>Integer</td>
      <td>Partition number</td>
    </tr>
    <tr>
      <td><code>:offset</code></td>
      <td>Integer</td>
      <td>Resolved offset</td>
    </tr>
    <tr>
      <td><code>:timestamp</code></td>
      <td>Integer</td>
      <td>Message timestamp (<code>-1</code> when not applicable)</td>
    </tr>
    <tr>
      <td><code>:leader_epoch</code></td>
      <td>Integer, nil</td>
      <td>Partition leader epoch</td>
    </tr>
  </tbody>
</table>

!!! warning "Each Partition May Appear at Most Once Per Call"

    A single `read_partition_offsets` call accepts each partition at most once. If you need both `:earliest` and `:latest` for the same partition, make two separate calls.

### Admin API / Reading Partition Offsets / Accurate Lag on Transactional Topics (LSO)

`read_watermark_offsets` always returns the high-watermark (HWM), which includes messages from uncommitted or in-flight transactions. A `READ_COMMITTED` consumer never sees those messages, so lag calculated from the HWM is overstated on transactionally-produced topics. Use `isolation_level: Karafka::Admin::IsolationLevels::READ_COMMITTED` to get the **Last Stable Offset (LSO)** instead:

```ruby
results = Karafka::Admin.read_partition_offsets(
  { 'events' => [{ partition: 0, offset: :latest }] },
  isolation_level: Karafka::Admin::IsolationLevels::READ_COMMITTED
)

lso = results.first[:offset]
puts "Last Stable Offset: #{lso}"
```

### Admin API / Reading Partition Offsets / Finding the Offset at a Point in Time

Pass a Unix timestamp **in milliseconds** as the `:offset` value to find the first offset at or after that timestamp:

```ruby
one_hour_ago_ms = (Time.now.to_f * 1000 - 3_600_000).to_i

results = Karafka::Admin.read_partition_offsets(
  { 'events' => [{ partition: 0, offset: one_hour_ago_ms }] }
)

puts "First offset from one hour ago: #{results.first[:offset]}"
```

### Admin API / Reading Partition Offsets / Finding the Offset of the Message with the Highest Timestamp

Use `:max_timestamp` to retrieve the offset of the message with the highest timestamp in the partition:

```ruby
results = Karafka::Admin.read_partition_offsets(
  { 'events' => [{ partition: 0, offset: :max_timestamp }] }
)

puts "Max-timestamp offset: #{results.first[:offset]}, at: #{results.first[:timestamp]}"
```

### Admin API / Reading Partition Offsets / Consumer-Side Counterpart

`Karafka::Admin.read_partition_offsets` (above) runs against a dedicated admin connection. A separate, lower-level `read_partition_offsets` also exists on `Connection::Client`/`Connection::Proxy` (`connection/client.rb`, `connection/proxy.rb`), letting a running consumer query offsets on its own connection instead of opening a new one. It accepts the same `topic_partition_offsets` spec, forwards the consumer's own `isolation.level`, and is used internally by the Pro Iterator and lag-compensation features.

Its own docstring notes a caveat: the underlying batched `ListOffsets` call resolves `:latest` to the high watermark regardless of the isolation level (unlike the consumer's `#query_watermark_offsets`, which returns the last stable offset for a read_committed consumer). On a topic with an in-flight transaction, `:latest` therefore includes uncommitted messages a read_committed consumer would not otherwise see. Non-transactional topics are unaffected (LSO == HWM).

## Admin API / Reading Lags and Offsets of a Consumer Group

This functionality provides the means to track and understand the consumption progress of consumer groups across specific topics and partitions within your Kafka setup. It is crucial for monitoring the health and performance of your Kafka consumers, as it helps identify any delays or backlogs in processing messages.

Using the following method, you can get the lags and offsets for a specific consumer group across selected topics. This will provide insights into how far behind each partition is from the latest message.

If consumer groups are not specified, the method will default to monitoring all consumer groups defined in the Karafka routing configuration with active topics, according to the `active_topics_only` parameter. This ensures comprehensive coverage of your Karafka setup, providing visibility into all potentially active consumption patterns without explicit specification.

### Admin API / Reading Lags and Offsets of a Consumer Group / Reading Lags of All Active Routing Topics

To get the lag for all active topics in your Karafka setup, invoke `#read_lags_with_offsets` without any arguments. Karafka will automatically use the consumer groups and active topics defined in `karafka.rb`:

```ruby
lags_with_offsets = Karafka::Admin.read_lags_with_offsets

lags_with_offsets.each do |group_name, topics|
  puts "Consumer group: #{group_name}"

  topics.each do |topic_name, partitions|
    puts "  Topic: #{topic_name}"

    partitions.each do |partition, data|
      offset = data[:offset]
      lag = data[:lag]
      puts "    Partition: #{partition}, Offset: #{offset}, Lag: #{lag}"
    end
  end

  puts
end

# Consumer group: cg0
#   Topic: visits
#     Partition: 0, Offset: 204468, Lag: 6
#     Partition: 1, Offset: 180631, Lag: 0
#     Partition: 2, Offset: 181414, Lag: 5
#     Partition: 3, Offset: 181070, Lag: 8
#     Partition: 4, Offset: 180974, Lag: 4
#
# Consumer group: karafka_web
#   Topic: karafka_consumers_reports
#     Partition: 0, Offset: 48669, Lag: 2
```

### Admin API / Reading Lags and Offsets of a Consumer Group / Reading Lags of All Routing Topics

To fetch lags for all topics, including inactive ones, set `active_topics_only` to `false` when calling `Karafka::Admin.read_lags_with_offsets`:

```ruby
lags_with_offsets = Karafka::Admin.read_lags_with_offsets(active_topics_only: false)

lags_with_offsets.each do |group_name, topics|
  puts "Consumer group: #{group_name}"

  topics.each do |topic_name, partitions|
    puts "  Topic: #{topic_name}"

    partitions.each do |partition, data|
      offset = data[:offset]
      lag = data[:lag]
      puts "    Partition: #{partition}, Offset: #{offset}, Lag: #{lag}"
    end
  end

  puts
end

# Consumer group: cg0
#   Topic: visits
#     Partition: 0, Offset: 204569, Lag: 12
#     Partition: 1, Offset: 180736, Lag: 9
#     Partition: 2, Offset: 181511, Lag: 11
#     Partition: 3, Offset: 181163, Lag: 3
#     Partition: 4, Offset: 181075, Lag: 14
#
# Consumer group: karafka_web
#   Topic: karafka_consumers_reports
#     Partition: 0, Offset: 48689, Lag: 3
#   Topic: karafka_consumers_states
#     Partition: 0, Offset: -1, Lag: -1
#   Topic: karafka_consumers_metrics
#     Partition: 0, Offset: -1, Lag: -1
#   Topic: karafka_consumers_commands
#     Partition: 0, Offset: -1, Lag: -1
#   Topic: karafka_errors
#     Partition: 0, Offset: -1, Lag: -1
```

### Admin API / Reading Lags and Offsets of a Consumer Group / Reading Lags of Selected Consumer Groups and Topics

To get lags for a specific subset of data, you can explicitly define the consumer groups and topics you are interested in:

```ruby
# Specify the consumer group and the topics you are interested in
consumer_group_with_topics = {
  'example_consumer_group' => ['topic1', 'topic2']
}

# Fetch the lags and offsets
lags_with_offsets = Karafka::Admin.read_lags_with_offsets(consumer_group_with_topics)

lags_with_offsets.each do |group_name, topics|
  puts "Consumer group: #{group_name}"

  topics.each do |topic_name, partitions|
    puts "  Topic: #{topic_name}"

    partitions.each do |partition, data|
      offset = data[:offset]
      lag = data[:lag]
      puts "    Partition: #{partition}, Offset: #{offset}, Lag: #{lag}"
    end
  end

  puts
end

# Consumer group: example_consumer_group
#   Topic: topic1
#     Partition: 0, Offset: 204676, Lag: 3
#     Partition: 1, Offset: 180825, Lag: 3
#     Partition: 2, Offset: 181592, Lag: 4
#     Partition: 3, Offset: 181249, Lag: 7
#     Partition: 4, Offset: 181154, Lag: 6
#
#   Topic: topic2
#     Partition: 0, Offset: 4676, Lag: 7
#     Partition: 1, Offset: 825, Lag: 12
```

### Admin API / Reading Lags and Offsets of a Consumer Group / Edge Cases and Considerations

- **Non-existent Topics**: Returns an empty hash for non-existent topics to indicate the absence of data.

- **Unconsumed Topics**: Topics with no recorded consumption by the specified consumer group will show a lag and offset of -1 for each partition, highlighting inactivity.

- **Kafka-centric Lag Reporting**: The reported lags reflect Kafka's internal tracking rather than the consumer's acknowledged state, which may differ based on the consumer's internal processing.

## Admin API / Describing Consumer Groups

`Karafka::Admin.describe_consumer_groups` returns the state of each requested consumer group, its partition assignor, its coordinator broker, and its members with the partitions assigned to each of them. It describes all the consumer groups in one request and returns an array in the order of the group ids you provide. To describe one consumer group, take the first element:

```ruby
Karafka::Admin.describe_consumer_groups(['billing']).first

# {
#   group_id: 'billing',
#   state: :stable,
#   partition_assignor: 'range',
#   coordinator: { id: 1, host: '127.0.0.1', port: 9092, rack: nil },
#   members: [
#     {
#       member_id: 'rdkafka-959acf9d-9706-46f8-b127-c63861b351dd',
#       client_id: 'rdkafka',
#       group_instance_id: nil,
#       host: '/172.31.0.1',
#       assignment: { 'payments' => [0, 1, 2] },
#       target_assignment: nil
#     }
#   ]
# }
```

To describe many consumer groups, provide all their ids:

```ruby
Karafka::Admin.describe_consumer_groups(%w[billing orders-processor]).each do |group|
  puts "#{group[:group_id]}: #{group[:state]} (#{group[:members].size} members)"
end
```

`Karafka::Admin::ConsumerGroups.describe` accepts one or more group ids as arguments and always returns an array, also when you describe one consumer group, for example `Karafka::Admin::ConsumerGroups.describe('billing').first`. On an Admin instance for a custom cluster, use `#describe_consumer_groups`.

Each consumer group description contains the following keys:

<table>
  <thead>
    <tr>
      <th>Key</th>
      <th>Description</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td><code>:group_id</code></td>
      <td>The consumer group id.</td>
    </tr>
    <tr>
      <td><code>:state</code></td>
      <td>The consumer group state as a symbol: <code>:unknown</code>, <code>:preparing_rebalance</code>, <code>:completing_rebalance</code>, <code>:stable</code>, <code>:dead</code>, or <code>:empty</code>.</td>
    </tr>
    <tr>
      <td><code>:partition_assignor</code></td>
      <td>The partition assignor in use, for example <code>"range"</code>, or <code>nil</code>.</td>
    </tr>
    <tr>
      <td><code>:coordinator</code></td>
      <td>The coordinator broker of the consumer group as <code>{ id:, host:, port:, rack: }</code>, or <code>nil</code>.</td>
    </tr>
    <tr>
      <td><code>:members</code></td>
      <td>The current members of the consumer group. Each member has <code>:member_id</code>, <code>:client_id</code>, <code>:group_instance_id</code>, <code>:host</code>, <code>:assignment</code>, and <code>:target_assignment</code>.</td>
    </tr>
  </tbody>
</table>

The `:assignment` of a member maps each topic name to an array of partitions. The `:target_assignment` has the same format and lists the partitions the member moves to during a rebalance. Only [KIP-848](https://karafka.io/docs/Kafka-New-Rebalance-Protocol.md) consumer groups have it. For other consumer groups, it is `nil`.

- **Unknown Consumer Groups**: A consumer group id that the cluster does not know is not an error. Kafka reports it in the `:dead` state with no members.

- **Stopped Consumer Groups**: After a clean shutdown, a consumer group keeps its offsets but has no live members. It is reported in the `:empty` state with no members.

- **Coordinator Not Ready**: On a fresh cluster, the consumer group coordinator may not be available yet. librdkafka does not retry this, so Karafka retries the `not_coordinator`, `coordinator_load_in_progress`, and `coordinator_not_available` errors for up to `admin.max_retries_duration`. When this time runs out, and on any other error, `Rdkafka::RdkafkaError` is raised.

## Admin API / Renaming a Consumer Group

!!! warning "Never Rename Active Consumer Groups"

    This method should **not** be used on actively running consumer groups, as it involves creating a temporary consumer to handle offset migration. Running this operation on active groups may cause unexpected behavior.

The `.rename_consumer_group` method in Karafka Admin API allows you to rename an existing consumer group while preserving its offsets for specific topics. This method is beneficial when reorganizing or consolidating consumer group names without losing track of the consumption state.

```ruby
Karafka::Admin.rename_consumer_group(
  'old_group_name',
  'new_group_name',
  ['topic1', 'topic2']
)
```

When using `rename_consumer_group`, the method ensures that offsets from the old consumer group are transferred to the new one, maintaining continuity in message consumption. You need to specify which topics should have their offsets migrated during the rename, giving you control over the process. By default, the original consumer group is deleted after the rename, but you can retain it by setting `delete_previous` to `false`.

!!! tip "Offset Merger with Existing Consumer Groups"

    If the new consumer group already exists, the offsets from the old group will be merged into it. This may result in the continuation of message processing from the combined offsets, so plan accordingly.

## Admin API / Copying a Consumer Group

!!! warning "Never Copy Active Consumer Groups"

    This method should **not** be used on actively running consumer groups, as it involves creating a temporary consumer to handle offset migration. Running this operation on active groups may cause unexpected behavior.

The `#copy_consumer_group` method in Karafka Admin API allows you to copy offsets from an existing consumer group to another while preserving its consumption state for specific topics. This functionality is useful when creating a duplicate consumer group with the same consumption progress as an existing one.

```ruby
Karafka::Admin.copy_consumer_group(
  'source_group_name',
  'target_group_name',
  ['topic1', 'topic2']
)
```

When using `#copy_consumer_group`, the method ensures that offsets from the source consumer group are transferred to the target one, maintaining continuity in message consumption. You need to specify which topics should have their offsets copied during the process, giving you control over what gets migrated.

!!! tip "Offset Merger with Existing Consumer Groups"

    If the target consumer group already exists, the offsets from the source group will be merged into it. This may result in the continuation of message processing from the combined offsets, so plan accordingly.

The method returns `false` only if the source consumer group does not exist at all (no lag data for any of the specified topics). If the topics exist but the source group has no committed offsets for them, `copy_consumer_group` still returns `true`, even though there is nothing to actually copy.

This functionality is particularly useful for:

- Creating backup consumer groups before making significant changes
- Testing new consumer configurations with the same consumption progress
- Setting up disaster recovery scenarios

Unlike `#rename_consumer_group`, this method preserves the source consumer group, allowing both groups to exist simultaneously.

## Admin API / Deleting a Consumer Group

!!! warning "Never Delete Active Consumer Groups"

    This method should only be used for consumer groups **not** actively used. Deleting a consumer group that is currently in use (running) can lead to data loss, inconsistencies, or unexpected behavior in your Kafka cluster.

The `Karafka::Admin.delete_consumer_group` method is designed to remove an existing consumer group from a Kafka cluster. This method effectively disbands the specified group, removing its metadata and state from the Kafka cluster.

To use it, all you need to do is to provide your consumer group name:

```ruby
Karafka::Admin.delete_consumer_group('your_consumer_group_name')
```

## Admin API / Triggering a Consumer Group Rebalance

!!! note "Development and Operational Use Only"

    This API should be used only for development or operational purposes, not as part of regular application logic.

The `Karafka::Admin.trigger_rebalance` method forces a rebalance for a given consumer group by briefly joining and leaving it with a temporary "fake" consumer: joining triggers the first rebalance, and leaving (after a short wait) triggers a second one. The group does not need to be running for this to work, but if it is, its consumers will experience both rebalances.

```ruby
Karafka::Admin.trigger_rebalance('my-group')
```

Topics are always detected from the routing configuration, and consumer settings are taken from the first topic in the group to stay consistent with the actual consumer configuration. `Karafka::Errors::InvalidConfigurationError` is raised if the group is not found in the routing or has no topics.

!!! note "Behavior Depends on the Rebalance Protocol"

    For cooperative-sticky or KIP-848-based rebalancing, there may be no immediate visible reaction to the trigger, as these protocols allow incremental partition reassignments without stopping all consumers.

## Admin API / Changing an Offset of a Consumer Group

!!! warning "Never Alter Active Consumer Groups"

    This method should only be used for consumer groups **not** actively used. Altering a consumer group that is currently in use (running) can lead to data loss, inconsistencies, or unexpected behavior in your Kafka cluster.

This method allows you to modify the offset for a specific topic within a consumer group, effectively controlling where the group starts consuming messages from within the topic.

The `Karafka::Admin.seek_consumer_group` method takes two primary arguments:

the name of the consumer group
a topic-partition hash with topics, partitions, and offsets where the consumer group should be moved

When invoked, it changes the current offset of the specified consumer group for the given topics to the new offsets provided. This method is beneficial when you need to reprocess or skip specific messages due to various operational requirements.

### Admin API / Changing an Offset of a Consumer Group / Changing an Offset for All Partitions

You can specify a hash mapping topics to integer offsets, allowing the adjustment of offsets for entire partitions within those topics:

```ruby
Karafka::Admin.seek_consumer_group(
  'my_consumer_group',
  {
    # move offset to 0 on all partitions of this topic
    'my_topic1' => 0,
    # move offset to 1000 on all partitions of this topic
    'my_other_topic' => 1000
  }

)
```

### Admin API / Changing an Offset of a Consumer Group / Changing an Offset for a Particular Partition

Alternatively, you can always move offset on specific partitions by specifying them directly:

```ruby
Karafka::Admin.seek_consumer_group(
  'my_consumer_group',
  {
    # move offset to 0 on partition 0 and to 10 on partition 1
    'my_topic1' => { 0 => 0, 1 => 10 },
    # move offset to 10 000 on partition 5 and to 50 000 on partition 20
    'my_other_topic' => { 5 => 10_000, 20 => 50_000 }
  }
)
```

### Admin API / Changing an Offset of a Consumer Group / Changing an Offset to a Time-Based Location

`.seek_consumer_group` method also accepts time references as offsets, allowing for precise time-based location seeking. Karafka automatically locates the matching offsets for the specified times and moves the consumer group position to this location.

```ruby
now = Time.now

Karafka::Admin.seek_consumer_group(
  'my_consumer_group',
  {
    # move offset back by 1 hour for partition 0, and 2 hours for partition 5
    'my_topic1' => { 0 => now - 60 * 60, 5 => now - 60 * 60 * 2 },
    # move offset back by five minutes for all the partitions of the given topic
    'my_other_topic' => now - 60 * 5
  }
)
```

### Admin API / Changing an Offset of a Consumer Group / Changing an Offset to Earliest or Latest

Adjusting consumer group offsets to "earliest" or "latest" helps control where consumption begins within a topic. However, these terms can be misleading, particularly with compacted topics where the "earliest" offset may not be zero due to log cleanup and message expiration. The "earliest" setting allows consumers to start from the oldest available message, ensuring comprehensive data coverage.

Conversely, the "latest" offset refers to the end of the log, indicating the point where new messages will be appended. Setting the offset to "latest" means consumption will start with new messages arriving post-adjustment, which is ideal for applications focused on real-time data.

The `Karafka::Admin.seek_consumer_group` method facilitates easy adjustment of offsets to either "earliest" or "latest", applicable across entire topics or specific partitions:

```ruby
# Adjust offsets for specific partitions
Karafka::Admin.seek_consumer_group(
  'your_consumer_group',
  {
    'your_topic' => {
      0 => :latest,    # Start reading new messages from partition 0
      1 => :earliest  # Start from the oldest message in partition 1
    }
  }
)

# Adjust all partitions of a topic to the earliest or latest
Karafka::Admin.seek_consumer_group(
  'your_consumer_group',
  { 'your_topic' => :earliest }  # Start all partitions from the oldest available message
)

Karafka::Admin.seek_consumer_group(
  'your_consumer_group',
  { 'your_topic' => :latest }  # Start consuming new messages across all partitions
)
```

Using `:earliest` and `:latest` is vital for managing consumer behavior, ensuring flexibility in data consumption strategies according to specific needs.

## Admin API / See Also

- [Admin Acls API](https://karafka.io/docs/Infrastructure-Admin-Acls-API.md) - Manage Kafka ACLs programmatically
- [Admin Configs API](https://karafka.io/docs/Infrastructure-Admin-Configs-API.md) - Configure topics and brokers
- [Admin Replication API](https://karafka.io/docs/Infrastructure-Admin-Replication-API.md) - Plan replication factor changes
- [Declarative Topics](https://karafka.io/docs/Infrastructure-Declarative-Topics.md) - Higher-level topic management approach


---

*Last modified: 2026-10-16 08:28:18*