Class: Ktl::Topic
Instance Method Summary collapse
- #add_partitions(*names) ⇒ Object
- #alter(regexp) ⇒ Object
- #create(*names) ⇒ Object
- #delete(regexp) ⇒ Object
- #describe(regexp = nil) ⇒ Object
- #list ⇒ Object
Instance Method Details
#add_partitions(*names) ⇒ Object
74 75 76 77 78 79 80 81 82 83 84 85 86 |
# File 'lib/ktl/topic.rb', line 74 def add_partitions(*names) with_zk_client do |zk_client| names.each do |name| opts = .merge(alter: nil, topic: name) = Kafka::Admin.(opts) logger.warn %(if "#{name}" uses keyed messages, the partition logic or ordering of the messages will be affected) silence_scala do Kafka::Admin::TopicCommand.alter_topic(zk_client.raw_client, ) end logger.info %(increased partitions to #{.partitions} for "#{name}") end end end |
#alter(regexp) ⇒ Object
106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 |
# File 'lib/ktl/topic.rb', line 106 def alter(regexp) with_zk_client do |zk_client| opts = {zookeeper: .zookeeper, topic: regexp} opts[:config] = .add.dup unless .add.empty? opts[:delete_config] = .remove.dup unless .remove.empty? if opts[:config] || opts[:delete_config] = Kafka::Admin.(opts) silence_scala do Kafka::Admin::TopicCommand.alter_topic(zk_client.raw_client, ) end logger.info %(updated configuration for topics matching "#{regexp}") else raise ArgumentError, 'missing --add or --remove option' end end end |
#create(*names) ⇒ Object
40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 |
# File 'lib/ktl/topic.rb', line 40 def create(*names) with_zk_client do |zk_client| names.each do |name| opts = .merge(create: nil, topic: name) if .rack_aware_allocation || .rendezvous_allocation plan_factory = if .rack_aware_allocation RackAwareShufflePlan else RendezvousShufflePlan end plan = plan_factory.new(zk_client, replication_factor: .replication_factor.to_i) zk_utils = Kafka::Utils::ZkUtils.new(nil, nil, false) opts.delete(:rack_aware_allocation) opts.delete(:rendezvous_allocation) plan = plan.generate_for_new_topic(name, .partitions.to_i) opts[:replica_assignment] = plan.map {|broker_list| broker_list.join(':')}.join(',') end = Kafka::Admin.(opts) silence_scala do Kafka::Admin::TopicCommand.create_topic(zk_client.raw_client, ) end = %(created topic "#{name}" with #{opts[:partitions]} partition(s)) << %(, and replication factor #{opts[:replication_factor]}) << %(, with replica assignment: #{opts[:replica_assignment]}) if opts[:replica_assignment] << %(, with config: #{opts[:config]}) unless opts[:config].empty? logger.info() end end end |
#delete(regexp) ⇒ Object
90 91 92 93 94 95 96 97 98 99 100 |
# File 'lib/ktl/topic.rb', line 90 def delete(regexp) with_zk_client do |zk_client| topics = zk_client.all_topics topics = topics.filter { |t| !!t.match(regexp) } logger.info %(about to mark #{topics.size} topics for deletion) topics.foreach do |topic| Kafka::Utils.delete_topic(zk_client.raw_client, topic) logger.debug %(successfully marked "#{topic}" for deletion) end end end |
#describe(regexp = nil) ⇒ Object
19 20 21 22 23 24 25 26 27 28 29 |
# File 'lib/ktl/topic.rb', line 19 def describe(regexp=nil) with_zk_client do |zk_client| opts = {describe: nil} opts[:topic] = regexp if regexp opts[:topics_with_overrides] = nil if .with_overrides? opts[:unavailable_partitions] = nil if .unavailable? opts[:under_replicated_partitions] = nil if .under_replicated? = Kafka::Admin.(opts) Kafka::Admin::TopicCommand.describe_topic(zk_client.raw_client, ) end end |