Class: KafkaCli::KafkaClient

Inherits:
Object
  • Object
show all
Defined in:
lib/kafka_cli/kafka_client.rb

Constant Summary collapse

DEFAULT_RESULT_WAIT = 
30

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(conf:) ⇒ KafkaClient

Returns a new instance of KafkaClient.



11
12
13
 
# File 'lib/kafka_cli/kafka_client.rb', line 11

def initialize(conf:)
  @conf = conf
end

Instance Attribute Details

#conf ⇒ Object (readonly)

Returns the value of attribute conf.



9
10
11
 
# File 'lib/kafka_cli/kafka_client.rb', line 9

def conf
  @conf
end

Instance Method Details

#admin ⇒ Object 



19
20
21
 
# File 'lib/kafka_cli/kafka_client.rb', line 19

def admin
  @admin ||= new_admin
end

#brokers ⇒ Object 



49
50
51
52
53
54
 
# File 'lib/kafka_cli/kafka_client.rb', line 49

def brokers(&)
  return enum_for(:topics) unless block_given?

  metadata = admin.metadata
  metadata.brokers.each(&)
end

#client ⇒ Object 



23
24
25
 
# File 'lib/kafka_cli/kafka_client.rb', line 23

def client
  @client ||= Rdkafka::Config.new(client_config)
end

#client_config ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/kafka_cli/kafka_client.rb', line 27

def client_config
  @client_config ||= {
    'client.id' => conf.client_id,
    'bootstrap.servers' => conf.brokers,
    'security.protocol' => conf.brokers_auth_mode,
    'ssl.certificate.location' => conf.ssl_certificate,
    'ssl.key.location' => conf.ssl_key,
    'ssl.key.password' => conf.ssl_key_password,
    'ssl.ca.location' => conf.ssl_ca,
    'enable.ssl.certificate.verification' => conf.ssl_verify,
    'socket.timeout.ms' => conf.socket_timeout,
    'request.timeout.ms' => conf.request_timeout
  }.compact
end

#create_topic(topic_name:, partitions: -1,, replicas: -1,, max_wait: DEFAULT_RESULT_WAIT) ⇒ Object 



56
57
58
59
60
61
62
63
64
65
 
# File 'lib/kafka_cli/kafka_client.rb', line 56

def create_topic(topic_name:, partitions: -1, replicas: -1, max_wait: DEFAULT_RESULT_WAIT)
  result = admin.create_topic(topic_name, partitions, replicas)
  result.wait(max_wait_timeout: max_wait)
rescue Rdkafka::RdkafkaError => e
  if e.code == :topic_already_exists
    puts "Topic '#{topic_name}' already exists."
  else
    puts "Failed to create topic '#{topic_name}'; error is #{e}"
  end
end

#create_topics(topic_names:, partitions: -1,, replicas: -1,, max_wait: DEFAULT_RESULT_WAIT) ⇒ Object 



67
68
69
70
71
72
 
# File 'lib/kafka_cli/kafka_client.rb', line 67

def create_topics(topic_names:, partitions: -1, replicas: -1, max_wait: DEFAULT_RESULT_WAIT)
  topic_names.each do |topic_name|
    create_topic(topic_name: topic_name, partitions: partitions, replicas: replicas,
                 max_wait: max_wait)
  end
end

#delete_topic(topic_name:, max_wait: DEFAULT_RESULT_WAIT) ⇒ Object 



74
75
76
77
78
79
80
81
82
83
 
# File 'lib/kafka_cli/kafka_client.rb', line 74

def delete_topic(topic_name:, max_wait: DEFAULT_RESULT_WAIT)
  result = admin.delete_topic(topic_name)
  result.wait(max_wait_timeout: max_wait)
rescue Rdkafka::RdkafkaError => e
  if e.code == :unknown_topic_or_part
    puts "Topic '#{topic_name}' does not exist or was already deleted."
  else
    puts "Failed to delete topic '#{topic_name}'; error is #{e}"
  end
end

#delete_topics(topic_names:, max_wait: DEFAULT_RESULT_WAIT) ⇒ Object 



85
86
87
88
89
 
# File 'lib/kafka_cli/kafka_client.rb', line 85

def delete_topics(topic_names:, max_wait: DEFAULT_RESULT_WAIT)
  topic_names.each do |topic_name|
    delete_topic(topic_name: topic_name, max_wait: max_wait)
  end
end

#new_admin ⇒ Object 



15
16
17
 
# File 'lib/kafka_cli/kafka_client.rb', line 15

def new_admin
  client.admin
end

#topics ⇒ Object 



42
43
44
45
46
47
 
# File 'lib/kafka_cli/kafka_client.rb', line 42

def topics(&)
  return enum_for(:topics) unless block_given?

  metadata = admin.metadata
  metadata.topics.each(&)
end