Class: Kafka::Admin

Inherits:
Object
  • Object
show all
Defined in:
lib/kafka/admin.rb

Overview

Admin provides a client for accessing the rdkafka Admin API to make changes to the cluster. The API provides was to create topics, delete topics, add new partitions for a topic, and manage configs

Instance Method Summary collapse

Constructor Details

#initialize(config = nil) ⇒ Admin

Create a new Admin client for accessing the librdkafka Admin API.

Parameters:



14
15
16
17
 
# File 'lib/kafka/admin.rb', line 14

def initialize(config = nil)
  # Wrap a Producer since it appears to allocate the fewest resources.
  @client = ::Kafka::FFI::Producer.new(config)
end

Instance Method Details

#create_topic(name, partitions, replication_factor, wait: true, validate: false, timeout: nil) ⇒ nil, TopicResult

Create a topic with the given name, number of partitions, and number of replicas per partition (replication factor). Total number of partitions will be partitions x replication_factor.

Parameters:

  • name (String) —

    Name of the topic to create

  • partitions (Integer) —

    Number of partitions the topic will have

  • replication_factor (Integer) —

    Number of replicas per partition to have in the cluster.

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

    Wait up to timeout milliseconds for topic creation to propogate to the cluster before returning.

  • validate (Boolean) (defaults to: false) —

    Only validate the request

  • timeout (Integer) (defaults to: nil) —

    Time to wait in milliseconds for each operation to complete. Total request execution time may be longer than timeout due to multiple operations being done. Defaults to ‘socket.timeout.ms` config setting.

Returns:

  • (nil) —

    Create timed out

  • (TopicResult) —

    Response from the cluster with details about if the topic was created or any errors.



39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/kafka/admin.rb', line 39

def create_topic(name, partitions, replication_factor, wait: true, validate: false, timeout: nil)
  req = ::Kafka::FFI::Admin::NewTopic.new(name, partitions, replication_factor)
  opts = new_options(:create_topics, wait: wait, validate: validate, timeout: timeout)

  res = @client.create_topics(req, options: opts)
  if res
    res[0]
  end
ensure
  opts.destroy
  req.destroy
end

#delete_topic(name, wait: true, validate: false, timeout: nil) ⇒ nil, TopicResult

Delete the topic with the given name

Parameters:

  • name (String) —

    Name of the topic to delete

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

    Wait up to timeout milliseconds for topic creation to propogate to the cluster before returning.

  • validate (Boolean) (defaults to: false) —

    Only validate the request

  • timeout (Integer) (defaults to: nil) —

    Time to wait in milliseconds for each operation to complete. Total request execution time may be longer than timeout due to multiple operations being done. Defaults to ‘socket.timeout.ms` config setting.

Returns:

  • (nil) —

    Delete timed out

  • (TopicResult) —

    Response from the cluster with details about the deletion or any errors.



67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/kafka/admin.rb', line 67

def delete_topic(name, wait: true, validate: false, timeout: nil)
  req = ::Kafka::FFI::Admin::DeleteTopic.new(name)
  opts = new_options(:create_topics, wait: wait, validate: validate, timeout: timeout)

  res = @client.delete_topics(req, options: opts)
  if res
    res[0]
  end
ensure
  opts.destroy
  req.destroy
end

#describe_config(type, name, wait: true, validate: false, timeout: nil) ⇒ ConfigResource

Get current config settings for the resource.

Examples:

Get configuration for a topic

describe_config(:topic, "events")

Parameters:

  • type (:broker, :topic, :group) —

    Type of resource

  • name (String) —

    Name of the resource

Returns:

  • (ConfigResource) 


89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/kafka/admin.rb', line 89

def describe_config(type, name, wait: true, validate: false, timeout: nil)
  req = ::Kafka::FFI::Admin::ConfigResource.new(type, name)
  opts = new_options(:create_topics, wait: wait, validate: validate, timeout: timeout)

  res = @client.describe_configs(req, options: opts)
  if res
    res[0]
  end
ensure
  opts.destroy
  req.destroy
end

#destroy ⇒ Object Also known as: close

Destroy the Client, releasing all used resources back to the system. It is the application’s responsbility to call #destroy when done with the client.



114
115
116
 
# File 'lib/kafka/admin.rb', line 114

def destroy
  @client.destroy
end

#metadata(local_only: false, topic: nil, timeout: 1000) ⇒ Metadata

Retrieve metadata for the cluster

Returns:

  • (Metadata)

See Also:



107
108
109
 
# File 'lib/kafka/admin.rb', line 107

def metadata(local_only: false, topic: nil, timeout: 1000)
  @client.metadata(local_only: local_only, topic: topic, timeout: timeout)
end