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 github.com/confluentinc/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
Creates an admin instance with this configuration.
-
#consumer ⇒ Consumer
Creates a consumer with this configuration.
-
#consumer_poll_set=(poll_set) ⇒ Object
Should we use a single queue for the underlying consumer and events.
-
#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.
112 113 114 115 116 |
# File 'lib/rdkafka/config.rb', line 112 def initialize(config_hash = {}) @config_hash = DEFAULT_CONFIG.merge(config_hash) @consumer_rebalance_listener = nil @consumer_poll_set = true end |
Class Method Details
.error_callback ⇒ Proc?
Returns the current error callback, by default this is nil.
86 87 88 |
# File 'lib/rdkafka/config.rb', line 86 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.
78 79 80 81 |
# File 'lib/rdkafka/config.rb', line 78 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.
38 39 40 |
# File 'lib/rdkafka/config.rb', line 38 def self.log_queue @@log_queue end |
.logger ⇒ Logger
Returns the current logger, by default this is a logger to stdout.
29 30 31 |
# File 'lib/rdkafka/config.rb', line 29 def self.logger @@logger end |
.logger=(logger) ⇒ nil
Set the logger that will be used for all logging output by this library.
47 48 49 50 |
# File 'lib/rdkafka/config.rb', line 47 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.
67 68 69 |
# File 'lib/rdkafka/config.rb', line 67 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: github.com/confluentinc/librdkafka/blob/master/STATISTICS.md
59 60 61 62 |
# File 'lib/rdkafka/config.rb', line 59 def self.statistics_callback=(callback) raise TypeError.new("Callback has to be callable") unless callback.respond_to?(:call) || callback == nil @@statistics_callback = callback end |
Instance Method Details
#[](key) ⇒ String?
Get a config option with the specified key
133 134 135 |
# File 'lib/rdkafka/config.rb', line 133 def [](key) @config_hash[key] end |
#[]=(key, value) ⇒ nil
Set a config option.
124 125 126 |
# File 'lib/rdkafka/config.rb', line 124 def []=(key, value) @config_hash[key] = value end |
#admin ⇒ Admin
Creates an admin instance with this configuration.
224 225 226 227 228 229 230 231 232 233 234 235 |
# File 'lib/rdkafka/config.rb', line 224 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, opaque: opaque ) ) end |
#consumer ⇒ Consumer
Creates a consumer with this configuration.
166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 |
# File 'lib/rdkafka/config.rb', line 166 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 queue Rdkafka::Bindings.rd_kafka_poll_set_consumer(kafka) if @consumer_poll_set # Return consumer with Kafka client Rdkafka::Consumer.new( Rdkafka::NativeKafka.new( kafka, run_polling_thread: false, opaque: opaque ) ) end |
#consumer_poll_set=(poll_set) ⇒ Object
Should we use a single queue for the underlying consumer and events.
This is an advanced API that allows for more granular control of the polling process. When this value is set to false
(true
by defualt), there will be two queues that need to be polled: - main librdkafka queue for events - consumer queue with messages and rebalances
It is recommended to use the defaults and only set it to false
in advance multi-threaded and complex cases where granular events handling control is needed.
156 157 158 |
# File 'lib/rdkafka/config.rb', line 156 def consumer_poll_set=(poll_set) @consumer_poll_set = poll_set end |
#consumer_rebalance_listener=(listener) ⇒ Object
Get notifications on partition assignment/revocation for the subscribed topics
140 141 142 |
# File 'lib/rdkafka/config.rb', line 140 def consumer_rebalance_listener=(listener) @consumer_rebalance_listener = listener end |
#producer ⇒ Producer
Create a producer with this configuration.
197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 |
# File 'lib/rdkafka/config.rb', line 197 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, opaque: opaque ), partitioner_name ).tap do |producer| opaque.producer = producer end end |