Class: Rdkafka::Config
- Inherits:
-
Object
- Object
- Rdkafka::Config
- Defined in:
- lib/rdkafka/config.rb
Overview
Configuration for a Kafka consumer or producer. You can create an instance and use the consumer and producer methods to create a client. Documentation of the available configuration options is available on https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md.
Defined Under Namespace
Classes: ClientCreationError, ConfigError, NoLoggerError
Constant Summary collapse
- DEFAULT_CONFIG =
Default config that can be overwritten.
{ # Request api version so advanced features work :"api.version.request" => true }.freeze
- REQUIRED_CONFIG =
Required config that cannot be overwritten.
{ # Enable log queues so we get callbacks in our own Ruby threads :"log.queue" => true }.freeze
Class Method Summary collapse
-
.error_callback ⇒ Proc?
Returns the current error callback, by default this is nil.
-
.error_callback=(callback) ⇒ nil
Set a callback that will be called every time the underlying client emits an error.
-
.log_queue ⇒ Queue
Returns a queue whose contents will be passed to the configured logger.
-
.logger ⇒ Logger
Returns the current logger, by default this is a logger to stdout.
-
.logger=(logger) ⇒ nil
Set the logger that will be used for all logging output by this library.
-
.statistics_callback ⇒ Proc?
Returns the current statistics callback, by default this is nil.
-
.statistics_callback=(callback) ⇒ nil
Set a callback that will be called every time the underlying client emits statistics.
Instance Method Summary collapse
-
#[](key) ⇒ String?
Get a config option with the specified key.
-
#[]=(key, value) ⇒ nil
Set a config option.
-
#admin ⇒ Admin
Create an admin instance with this configuration.
-
#consumer ⇒ Consumer
Create a consumer with this configuration.
-
#consumer_rebalance_listener=(listener) ⇒ Object
Get notifications on partition assignment/revocation for the subscribed topics.
-
#initialize(config_hash = {}) ⇒ Config
constructor
Returns a new config with the provided options which are merged with DEFAULT_CONFIG.
-
#producer ⇒ Producer
Create a producer with this configuration.
Constructor Details
#initialize(config_hash = {}) ⇒ Config
Returns a new config with the provided options which are merged with DEFAULT_CONFIG.
114 115 116 117 |
# File 'lib/rdkafka/config.rb', line 114 def initialize(config_hash = {}) @config_hash = DEFAULT_CONFIG.merge(config_hash) @consumer_rebalance_listener = nil end |
Class Method Details
.error_callback ⇒ Proc?
Returns the current error callback, by default this is nil.
88 89 90 |
# File 'lib/rdkafka/config.rb', line 88 def self.error_callback @@error_callback end |
.error_callback=(callback) ⇒ nil
Set a callback that will be called every time the underlying client emits an error. If this callback is not set, global errors such as brokers becoming unavailable will only be sent to the logger, as defined by librdkafka. The callback is called with an instance of RdKafka::Error.
80 81 82 83 |
# File 'lib/rdkafka/config.rb', line 80 def self.error_callback=(callback) raise TypeError.new("Callback has to be callable") unless callback.respond_to?(:call) @@error_callback = callback end |
.log_queue ⇒ Queue
Returns a queue whose contents will be passed to the configured logger. Each entry should follow the format [Logger::Severity, String]. The benefit over calling the logger directly is that this is safe to use from trap contexts.
40 41 42 |
# File 'lib/rdkafka/config.rb', line 40 def self.log_queue @@log_queue end |
.logger ⇒ Logger
Returns the current logger, by default this is a logger to stdout.
31 32 33 |
# File 'lib/rdkafka/config.rb', line 31 def self.logger @@logger end |
.logger=(logger) ⇒ nil
Set the logger that will be used for all logging output by this library.
49 50 51 52 |
# File 'lib/rdkafka/config.rb', line 49 def self.logger=(logger) raise NoLoggerError if logger.nil? @@logger = logger end |
.statistics_callback ⇒ Proc?
Returns the current statistics callback, by default this is nil.
69 70 71 |
# File 'lib/rdkafka/config.rb', line 69 def self.statistics_callback @@statistics_callback end |
.statistics_callback=(callback) ⇒ nil
Set a callback that will be called every time the underlying client emits statistics.
You can configure if and how often this happens using statistics.interval.ms
.
The callback is called with a hash that's documented here: https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md
61 62 63 64 |
# File 'lib/rdkafka/config.rb', line 61 def self.statistics_callback=(callback) raise TypeError.new("Callback has to be callable") unless callback.respond_to?(:call) @@statistics_callback = callback end |
Instance Method Details
#[](key) ⇒ String?
Get a config option with the specified key
134 135 136 |
# File 'lib/rdkafka/config.rb', line 134 def [](key) @config_hash[key] end |
#[]=(key, value) ⇒ nil
Set a config option.
125 126 127 |
# File 'lib/rdkafka/config.rb', line 125 def []=(key, value) @config_hash[key] = value end |
#admin ⇒ Admin
Create an admin instance with this configuration.
196 197 198 199 200 201 |
# File 'lib/rdkafka/config.rb', line 196 def admin opaque = Opaque.new config = native_config(opaque) Rdkafka::Bindings.rd_kafka_conf_set_background_event_cb(config, Rdkafka::Callbacks::BackgroundEventCallbackFunction) Rdkafka::Admin.new(Rdkafka::NativeKafka.new(native_kafka(config, :rd_kafka_producer), run_polling_thread: true)) end |
#consumer ⇒ Consumer
Create a consumer with this configuration.
151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
# File 'lib/rdkafka/config.rb', line 151 def consumer opaque = Opaque.new config = native_config(opaque) if @consumer_rebalance_listener opaque.consumer_rebalance_listener = @consumer_rebalance_listener Rdkafka::Bindings.rd_kafka_conf_set_rebalance_cb(config, Rdkafka::Bindings::RebalanceCallback) end # Create native client kafka = native_kafka(config, :rd_kafka_consumer) # Redirect the main queue to the consumer Rdkafka::Bindings.rd_kafka_poll_set_consumer(kafka) # Return consumer with Kafka client Rdkafka::Consumer.new(Rdkafka::NativeKafka.new(kafka, run_polling_thread: false)) end |
#consumer_rebalance_listener=(listener) ⇒ Object
Get notifications on partition assignment/revocation for the subscribed topics
141 142 143 |
# File 'lib/rdkafka/config.rb', line 141 def consumer_rebalance_listener=(listener) @consumer_rebalance_listener = listener end |
#producer ⇒ Producer
Create a producer with this configuration.
176 177 178 179 180 181 182 183 184 185 186 187 188 |
# File 'lib/rdkafka/config.rb', line 176 def producer # Create opaque opaque = Opaque.new # Create Kafka config config = native_config(opaque) # Set callback to receive delivery reports on config Rdkafka::Bindings.rd_kafka_conf_set_dr_msg_cb(config, Rdkafka::Callbacks::DeliveryCallbackFunction) # Return producer with Kafka client partitioner_name = self[:partitioner] || self["partitioner"] Rdkafka::Producer.new(Rdkafka::NativeKafka.new(native_kafka(config, :rd_kafka_producer), run_polling_thread: true), partitioner_name).tap do |producer| opaque.producer = producer end end |