Class: Karafka::Admin

Inherits:
Object
  • Object
show all
Extended by:
Core::Helpers::Time
Defined in:
lib/karafka/admin.rb,
lib/karafka/admin/acl.rb,
lib/karafka/admin/topics.rb,
lib/karafka/admin/configs.rb,
lib/karafka/admin/replication.rb,
lib/karafka/admin/configs/config.rb,
lib/karafka/admin/consumer_groups.rb,
lib/karafka/admin/configs/resource.rb,
lib/karafka/admin/isolation_levels.rb,
lib/karafka/admin/contracts/replication.rb

Overview

Note:

Each operation initializes its own dedicated instance as we want to ensure it is always closed. Since admin actions are not performed that often, that should be ok. When an external client is provided, it is reused instead and its lifecycle stays fully with its owner.

Note:

By default it uses the primary defined cluster. For multi-cluster operations, create an Admin instance with custom kafka configuration: Karafka::Admin.new(kafka: { 'bootstrap.servers': 'other:9092' })

Admin actions that we can perform via Karafka on our Kafka cluster

Direct Known Subclasses

Pro::Admin::Recovery

Defined Under Namespace

Modules: Contracts, IsolationLevels Classes: Acl, Configs, ConsumerGroups, Replication, Topics

Constant Summary collapse

Recovery =

We alias this for Pro users so we don't end up having two Admin namespaces from the end user perspective. This enhances the UX.

Karafka::Pro::Admin::Recovery

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(kafka: {}, external_client: nil) ⇒ Admin

Note:

Raw and proxied rdkafka instances are resolved once at construction, so the admin instance should not outlive them. Karafka::Connection::Client instances are resolved on each use instead, so the admin instance follows such a client across the underlying connection recovery resets.

Creates a new Admin instance

Examples:

Create admin for a different cluster

admin = Karafka::Admin.new(kafka: { 'bootstrap.servers': 'other-cluster:9092' })
admin.cluster_info

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'] })

Parameters:

  • kafka (Hash) (defaults to: {}) —

    custom kafka configuration to merge with app defaults. Useful for multi-cluster operations where you want to target a different cluster.

  • external_client (Object, nil) (defaults to: nil) —

    active rdkafka client (raw, wrapped with Karafka::Connection::Proxy or a Karafka::Connection::Client of a running consumer) on which admin operations should run, instead of each operation creating its own short-lived instance. Routing is capability based: rdkafka admin instances are used by admin-based operations (with_admin and everything built on top of it, e.g. cluster_info), any other external client is used by consumer-based operations (with_consumer and everything built on top of it). The lifecycle of an external client belongs fully to its owner: it is never configured, started or closed here and all operations run within its identity, including its group.id. This is a low-level internal API: the caller is responsible for providing a client capable of the invoked operations and for invoking only operations that are safe to run on a live client.



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/karafka/admin.rb', line 55

def initialize(kafka: {}, external_client: nil)
  @custom_kafka = kafka
  @external_client = nil
  @external_admin = false
  @external_consumer = false

  return unless external_client

  # A running consumer connection client is kept as-is and its current handle is resolved
  # on each use: such clients swap their underlying rdkafka instance across recovery
  # resets, so a handle resolved once at construction could go stale while the client
  # itself remains fully operational
  if external_client.is_a?(Karafka::Connection::Client)
    @external_client = external_client
    @external_consumer = true

    return
  end

  # Raw and proxied rdkafka instances are wrapped upfront - the proxy itself prevents
  # double wrapping
  @external_client = Karafka::Connection::Proxy.new(external_client)

  # Capability based routing resolved once upfront: rdkafka admin instances serve
  # admin-based operations, any other external client is assumed consumer capable
  if @external_client.wrapped.is_a?(Rdkafka::Admin)
    @external_admin = true
  else
    @external_consumer = true
  end
end

Class Method Details

.cluster_info ⇒ Rdkafka::Metadata

Returns cluster metadata info.

Returns:

  • (Rdkafka::Metadata) —

    cluster metadata info



278
279
280
 
# File 'lib/karafka/admin.rb', line 278

def cluster_info
  new.cluster_info
end

.copy_consumer_group(previous_name, new_name, topics) ⇒ Boolean

Takes consumer group and its topics and copies all the offsets to a new named group

Parameters:

  • previous_name (String) —

    old consumer group name

  • new_name (String) —

    new consumer group name

  • topics (Array<String>) —

    topics for which we want to migrate offsets during rename

Returns:

  • (Boolean) —

    true if anything was migrated, otherwise false

See Also:



174
175
176
 
# File 'lib/karafka/admin.rb', line 174

def copy_consumer_group(previous_name, new_name, topics)
  new.copy_consumer_group(previous_name, new_name, topics)
end

.create_partitions(name, partitions) ⇒ Object

Parameters:

  • name (String) —

    topic name

  • partitions (Integer) —

    total number of partitions we expect to end up with

See Also:



135
136
137
 
# File 'lib/karafka/admin.rb', line 135

def create_partitions(name, partitions)
  new.create_partitions(name, partitions)
end

.create_topic(name, partitions, replication_factor, topic_config = {}) ⇒ Object

Parameters:

  • name (String) —

    topic name

  • partitions (Integer) —

    number of partitions we expect

  • replication_factor (Integer) —

    number of replicas

  • topic_config (Hash) (defaults to: {}) —

    topic config details

See Also:



122
123
124
 
# File 'lib/karafka/admin.rb', line 122

def create_topic(name, partitions, replication_factor, topic_config = {})
  new.create_topic(name, partitions, replication_factor, topic_config)
end

.delete_consumer_group(group_id) ⇒ Object

Removes given group (if exists)

Parameters:

  • group_id (String) —

    group name

See Also:



201
202
203
 
# File 'lib/karafka/admin.rb', line 201

def delete_consumer_group(group_id)
  new.delete_consumer_group(group_id)
end

.delete_topic(name) ⇒ Object

Parameters:

  • name (String) —

    topic name

See Also:



128
129
130
 
# File 'lib/karafka/admin.rb', line 128

def delete_topic(name)
  new.delete_topic(name)
end

.plan_topic_replication(topic:, replication_factor:, brokers: nil) ⇒ Replication

Plans a replication factor increase for a topic that can be used with Kafka's reassignment tools. Since librdkafka does not support increasing replication factor directly, this method generates the necessary JSON and commands for manual execution.

Examples:

Plan replication increase with automatic broker distribution

plan = Karafka::Admin.plan_topic_replication(topic: 'events', replication_factor: 3)

# Review the plan
puts plan.summary

# Export JSON for Kafka's reassignment tools
plan.export_to_file('reassignment.json')

# Execute the plan (replace <KAFKA_BROKERS> with actual brokers)
system(plan.execution_commands[:execute].gsub('<KAFKA_BROKERS>', 'localhost:9092'))

Plan replication with manual broker placement - specify brokers

plan = Karafka::Admin.plan_topic_replication(
  topic: 'events',
  replication_factor: 3,
  brokers: {
    0 => [1, 2, 4],  # Partition 0 on brokers 1, 2, 4
    1 => [2, 3, 4],  # Partition 1 on brokers 2, 3, 4
    2 => [1, 3, 5]   # Partition 2 on brokers 1, 3, 5
  }
)

# The plan will use your exact broker specifications
puts plan.partitions_assignment
# => { 0=>[1, 2, 4], 1=>[2, 3, 4], 2=>[1, 3, 5] }

Parameters:

  • topic (String) —

    name of the topic to plan replication for

  • replication_factor (Integer) —

    target replication factor (must be higher than current)

  • brokers (Hash{Integer => Array<Integer>}) (defaults to: nil) —

    optional manual broker assignments per partition. Keys are partition IDs, values are arrays of broker IDs. If not provided, assignments distribution will happen automatically.

Returns:

  • (Replication) —

    plan object with JSON, commands, and instructions

See Also:



269
270
271
272
273
274
275
 
# File 'lib/karafka/admin.rb', line 269

def plan_topic_replication(topic:, replication_factor:, brokers: nil)
  new.plan_topic_replication(
    topic: topic,
    replication_factor: replication_factor,
    brokers: brokers
  )
end

.read_lags_with_offsets(groups_with_topics = {}, active_topics_only: true) ⇒ Hash{String => Hash{Integer => Hash{Integer => Object}}}

Reads lags and offsets for given topics in the context of groups defined in the routing

Parameters:

  • groups_with_topics (Hash{String => Array<String>}) (defaults to: {}) —

    hash with group names with array of topics to query per group inside

  • active_topics_only (Boolean) (defaults to: true) —

    if set to false, when we use routing topics, will select also topics that are marked as inactive in routing

Returns:

  • (Hash{String => Hash{Integer => Hash{Integer => Object}}}) —

    hash where the top level keys are the groups and values are hashes with topics and inside partitions with lags and offsets

See Also:



223
224
225
226
227
228
 
# File 'lib/karafka/admin.rb', line 223

def read_lags_with_offsets(groups_with_topics = {}, active_topics_only: true)
  new.read_lags_with_offsets(
    groups_with_topics,
    active_topics_only: active_topics_only
  )
end

.read_partition_offsets(topic_partition_offsets, isolation_level: nil) ⇒ Object

Parameters:

  • topic_partition_offsets (Hash{String => Array<Hash>}) —

    topics with partition specs

  • isolation_level (Integer, nil) (defaults to: nil) —

    optional isolation level constant

See Also:



149
150
151
 
# File 'lib/karafka/admin.rb', line 149

def read_partition_offsets(topic_partition_offsets, isolation_level: nil)
  new.read_partition_offsets(topic_partition_offsets, isolation_level: isolation_level)
end

.read_topic(name, partition, count, start_offset = -1,, settings = {}) ⇒ Object

Parameters:

  • name (String, Symbol) —

    topic name

  • partition (Integer) —

    partition

  • count (Integer) —

    how many messages we want to get at most

  • start_offset (Integer, Time) (defaults to: -1,) —

    offset from which we should start

  • settings (Hash) (defaults to: {}) —

    kafka extra settings

See Also:



113
114
115
 
# File 'lib/karafka/admin.rb', line 113

def read_topic(name, partition, count, start_offset = -1, settings = {})
  new.read_topic(name, partition, count, start_offset, settings)
end

.read_watermark_offsets(name_or_hash, partition = nil) ⇒ Object

Parameters:

  • name_or_hash (String, Symbol, Hash) —

    topic name or hash with topics and partitions

  • partition (Integer, nil) (defaults to: nil) —

    partition (nil when using hash format)

See Also:



142
143
144
 
# File 'lib/karafka/admin.rb', line 142

def read_watermark_offsets(name_or_hash, partition = nil)
  new.read_watermark_offsets(name_or_hash, partition)
end

.rename_consumer_group(previous_name, new_name, topics, delete_previous: true) ⇒ Boolean

Takes consumer group and its topics and migrates all the offsets to a new named group

Parameters:

  • previous_name (String) —

    old consumer group name

  • new_name (String) —

    new consumer group name

  • topics (Array<String>) —

    topics for which we want to migrate offsets during rename

  • delete_previous (Boolean) (defaults to: true) —

    should we delete previous consumer group after rename. Defaults to true.

Returns:

  • (Boolean) —

    true if rename (and optionally removal) was ok or false if there was nothing really to rename

See Also:



188
189
190
191
192
193
194
195
 
# File 'lib/karafka/admin.rb', line 188

def rename_consumer_group(previous_name, new_name, topics, delete_previous: true)
  new.rename_consumer_group(
    previous_name,
    new_name,
    topics,
    delete_previous: delete_previous
  )
end

.seek_consumer_group(group_id, topics_with_partitions_and_offsets) ⇒ Object

Parameters:

  • group_id (String) —

    id of the group for which we want to move the existing offset

  • topics_with_partitions_and_offsets (Hash) —

    Hash with list of topics and settings

See Also:



163
164
165
 
# File 'lib/karafka/admin.rb', line 163

def seek_consumer_group(group_id, topics_with_partitions_and_offsets)
  new.seek_consumer_group(group_id, topics_with_partitions_and_offsets)
end

.topic_info(topic_name) ⇒ Object

Parameters:

  • topic_name (String) —

    name of the topic we're interested in

See Also:



155
156
157
 
# File 'lib/karafka/admin.rb', line 155

def topic_info(topic_name)
  new.topic_info(topic_name)
end

.trigger_rebalance(group_id) ⇒ Object

Note:

This API should be used only for development.

Triggers a rebalance for the specified group

Parameters:

  • group_id (String) —

    group id to trigger rebalance for

See Also:



210
211
212
 
# File 'lib/karafka/admin.rb', line 210

def trigger_rebalance(group_id)
  new.trigger_rebalance(group_id)
end

.with_admin ⇒ Object

Creates admin instance and yields it. After usage it closes the admin instance



294
295
296
 
# File 'lib/karafka/admin.rb', line 294

def with_admin(&)
  new.with_admin(&)
end

.with_consumer(settings = {}) ⇒ Object

Note:

We always ship and yield a proxied consumer because admin API performance is not that relevant. That is, there are no high frequency calls that would have to be delegated

Creates consumer instance and yields it. After usage it closes the consumer instance This API can be used in other pieces of code and allows for low-level consumer usage

Parameters:

  • settings (Hash) (defaults to: {}) —

    extra settings to customize consumer



289
290
291
 
# File 'lib/karafka/admin.rb', line 289

def with_consumer(settings = {}, &)
  new.with_consumer(settings, &)
end

Instance Method Details

#close ⇒ Object

No-op close to normalize the API surface.

Each admin operation opens and closes its own dedicated rdkafka instance internally, while external clients are owned and closed by their providers, so there is nothing to release at the Karafka::Admin level right now. This method exists so that callers who hold an instance and call #close on it (matching the pattern of other closeable resources) do not raise NoMethodError.

In the future, Karafka::Admin is planned to be refactored to reuse a single owned rdkafka admin instance across multiple operations rather than creating and tearing one down per call. When that happens, this method will need to release that shared owned instance - though never an external client. The no-op is here now so that all callers are already written against the correct API and require no changes when the real implementation lands.



101
102
 
# File 'lib/karafka/admin.rb', line 101

def close
end

#cluster_info ⇒ Rdkafka::Metadata

Returns cluster metadata info.

Returns:

  • (Rdkafka::Metadata) —

    cluster metadata info



421
422
423
 
# File 'lib/karafka/admin.rb', line 421

def cluster_info
  with_admin(&:metadata)
end

#copy_consumer_group(previous_name, new_name, topics) ⇒ Object

Parameters:

  • previous_name (String) —

    old consumer group name

  • new_name (String) —

    new consumer group name

  • topics (Array<String>) —

    topics for which we want to copy offsets

See Also:



367
368
369
 
# File 'lib/karafka/admin.rb', line 367

def copy_consumer_group(previous_name, new_name, topics)
  consumer_groups_admin.copy(previous_name, new_name, topics)
end

#create_partitions(name, partitions) ⇒ Object

Parameters:

  • name (String) —

    topic name

  • partitions (Integer) —

    total number of partitions we expect to end up with

See Also:



329
330
331
 
# File 'lib/karafka/admin.rb', line 329

def create_partitions(name, partitions)
  topics_admin.create_partitions(name, partitions)
end

#create_topic(name, partitions, replication_factor, topic_config = {}) ⇒ Object

Parameters:

  • name (String) —

    topic name

  • partitions (Integer) —

    number of partitions for this topic

  • replication_factor (Integer) —

    number of replicas

  • topic_config (Hash) (defaults to: {}) —

    topic config details

See Also:



316
317
318
 
# File 'lib/karafka/admin.rb', line 316

def create_topic(name, partitions, replication_factor, topic_config = {})
  topics_admin.create(name, partitions, replication_factor, topic_config)
end

#delete_consumer_group(group_id) ⇒ Object

Parameters:

  • group_id (String) —

    group name

See Also:



387
388
389
 
# File 'lib/karafka/admin.rb', line 387

def delete_consumer_group(group_id)
  consumer_groups_admin.delete(group_id)
end

#delete_topic(name) ⇒ Object

Parameters:

  • name (String) —

    topic name

See Also:



322
323
324
 
# File 'lib/karafka/admin.rb', line 322

def delete_topic(name)
  topics_admin.delete(name)
end

#plan_topic_replication(topic:, replication_factor:, brokers: nil) ⇒ Object

Parameters:

  • topic (String) —

    topic name to plan replication for

  • replication_factor (Integer) —

    target replication factor

  • brokers (Hash, nil) (defaults to: nil) —

    optional manual broker assignments per partition

See Also:



412
413
414
415
416
417
418
 
# File 'lib/karafka/admin.rb', line 412

def plan_topic_replication(topic:, replication_factor:, brokers: nil)
  Replication.new(kafka: @custom_kafka, external_client: @external_client).plan(
    topic: topic,
    to: replication_factor,
    brokers: brokers
  )
end

#read_lags_with_offsets(groups_with_topics = {}, active_topics_only: true) ⇒ Object

Parameters:

  • groups_with_topics (Hash{String => Array<String>}) (defaults to: {}) —

    hash with group names with array of topics

  • active_topics_only (Boolean) (defaults to: true) —

    if set to false, will select also inactive topics

See Also:



401
402
403
404
405
406
 
# File 'lib/karafka/admin.rb', line 401

def read_lags_with_offsets(groups_with_topics = {}, active_topics_only: true)
  consumer_groups_admin.read_lags_with_offsets(
    groups_with_topics,
    active_topics_only: active_topics_only
  )
end

#read_partition_offsets(topic_partition_offsets, isolation_level: nil) ⇒ Object

Parameters:

  • topic_partition_offsets (Hash{String => Array<Hash>}) —

    topics with partition specs

  • isolation_level (Integer, nil) (defaults to: nil) —

    optional isolation level constant

See Also:



343
344
345
 
# File 'lib/karafka/admin.rb', line 343

def read_partition_offsets(topic_partition_offsets, isolation_level: nil)
  topics_admin.read_partition_offsets(topic_partition_offsets, isolation_level: isolation_level)
end

#read_topic(name, partition, count, start_offset = -1,, settings = {}) ⇒ Object

Parameters:

  • name (String, Symbol) —

    topic name

  • partition (Integer) —

    partition

  • count (Integer) —

    how many messages we want to get at most

  • start_offset (Integer, Time) (defaults to: -1,) —

    offset from which we should start

  • settings (Hash) (defaults to: {}) —

    kafka extra settings (optional)

See Also:



307
308
309
 
# File 'lib/karafka/admin.rb', line 307

def read_topic(name, partition, count, start_offset = -1, settings = {})
  topics_admin.read(name, partition, count, start_offset, settings)
end

#read_watermark_offsets(name_or_hash, partition = nil) ⇒ Object

Parameters:

  • name_or_hash (String, Symbol, Hash) —

    topic name or hash with topics and partitions

  • partition (Integer, nil) (defaults to: nil) —

    partition (nil when using hash format)

See Also:



336
337
338
 
# File 'lib/karafka/admin.rb', line 336

def read_watermark_offsets(name_or_hash, partition = nil)
  topics_admin.read_watermark_offsets(name_or_hash, partition)
end

#rename_consumer_group(previous_name, new_name, topics, delete_previous: true) ⇒ Object

Parameters:

  • previous_name (String) —

    old consumer group name

  • new_name (String) —

    new consumer group name

  • topics (Array<String>) —

    topics for which we want to migrate offsets

  • delete_previous (Boolean) (defaults to: true) —

    should we delete previous consumer group after rename

See Also:



376
377
378
379
380
381
382
383
 
# File 'lib/karafka/admin.rb', line 376

def rename_consumer_group(previous_name, new_name, topics, delete_previous: true)
  consumer_groups_admin.rename(
    previous_name,
    new_name,
    topics,
    delete_previous: delete_previous
  )
end

#seek_consumer_group(group_id, topics_with_partitions_and_offsets) ⇒ Object

Parameters:

  • group_id (String) —

    group for which we want to move offsets

  • topics_with_partitions_and_offsets (Hash) —

    hash with topics and settings

See Also:



356
357
358
359
360
361
 
# File 'lib/karafka/admin.rb', line 356

def seek_consumer_group(group_id, topics_with_partitions_and_offsets)
  consumer_groups_admin.seek(
    group_id,
    topics_with_partitions_and_offsets
  )
end

#topic_info(topic_name) ⇒ Object

Parameters:

  • topic_name (String) —

    name of the topic we're interested in

See Also:



349
350
351
 
# File 'lib/karafka/admin.rb', line 349

def topic_info(topic_name)
  topics_admin.info(topic_name)
end

#trigger_rebalance(group_id) ⇒ Object

Parameters:

  • group_id (String) —

    group id to trigger rebalance for

See Also:



393
394
395
 
# File 'lib/karafka/admin.rb', line 393

def trigger_rebalance(group_id)
  consumer_groups_admin.trigger_rebalance(group_id)
end

#with_admin ⇒ Object

Note:

When the external client is an rdkafka admin instance, it is yielded wrapped with Karafka::Connection::Proxy (which unwraps pre-proxied clients, so no double wrapping occurs) and its lifecycle is not managed here in any way, same as with #with_consumer. External clients of other types (consumers, producers) are not capable of admin operations, thus a dedicated admin instance is created for them as usual.

Creates admin instance and yields it. After usage it closes the admin instance



478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
 
# File 'lib/karafka/admin.rb', line 478

def with_admin
  return yield(external_client_proxy) if @external_admin

  bind_id = SecureRandom.uuid
  admin = nil

  begin
    admin = config(:producer, {}).admin(
      native_kafka_auto_start: false,
      native_kafka_poll_timeout_ms: self.class.poll_timeout
    )

    bind_oauth(bind_id, admin)

    admin.start
    proxy = Karafka::Connection::Proxy.new(admin)
    yield(proxy)
  ensure
    admin&.close

    unbind_oauth(bind_id)
  end
end

#with_consumer(settings = {}) ⇒ Object

Note:

We always ship and yield a proxied consumer because admin API performance is not that relevant. That is, there are no high frequency calls that would have to be delegated

Note:

When an external client is present, it is yielded wrapped with Karafka::Connection::Proxy (which unwraps pre-proxied clients, so no double wrapping occurs) and its lifecycle is not managed here in any way: no oauth binding, no start and no closing. settings are ignored as the external instance is already configured and operations run within its identity, including its group.id. External rdkafka admin instances are not used here as they are not capable of consumer operations - they are used by #with_admin instead.

Creates consumer instance and yields it. After usage it closes the consumer instance This API can be used in other pieces of code and allows for low-level consumer usage

Parameters:

  • settings (Hash) (defaults to: {}) —

    extra settings to customize consumer



440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
 
# File 'lib/karafka/admin.rb', line 440

def with_consumer(settings = {})
  return yield(external_client_proxy) if @external_consumer

  bind_id = SecureRandom.uuid
  consumer = nil

  begin
    consumer = config(:consumer, settings).consumer(native_kafka_auto_start: false)
    bind_oauth(bind_id, consumer)

    consumer.start
    proxy = Karafka::Connection::Proxy.new(consumer)
    yield(proxy)
  ensure
    # Always unsubscribe consumer just to be sure, that no metadata requests are running
    # when we close the consumer. This in theory should prevent from some race-conditions
    # that originate from librdkafka
    begin
      consumer&.unsubscribe
    # Ignore any errors and continue to close consumer despite them
    rescue Rdkafka::RdkafkaError
      nil
    end

    consumer&.close

    unbind_oauth(bind_id)
  end
end