Class: Sbmt::KafkaProducer::Config::Kafka

Inherits:
Dry::Struct
  • Object
show all
Defined in:
lib/sbmt/kafka_producer/config/kafka.rb

Constant Summary collapse

SERVERS_REGEXP =

srv1:port1,srv2:port2,…

/^[a-z\d.\-:]+(,[a-z\d.\-:]+)*$/.freeze
KAFKA_CONFIG_KEYS_REMAP =

github.com/karafka/waterdrop/blob/master/lib/waterdrop/config.rb

{
  servers: :"bootstrap.servers",
  connect_timeout: :"socket.connection.setup.timeout.ms",
  message_timeout: :"message.timeout.ms",
  ack_timeout: :"request.timeout.ms",
  retry_backoff: :"retry.backoff.ms",
  max_retries: :"message.send.max.retries",
  required_acks: :"request.required.acks"
}

Instance Method Summary collapse

Instance Method Details

#to_kafka_options ⇒ Object 



36
37
38
39
40
41
42
 
# File 'lib/sbmt/kafka_producer/config/kafka.rb', line 36

def to_kafka_options
  cfg = KAFKA_CONFIG_KEYS_REMAP.each_with_object({}) do |(key, kafka_key), hash|
    hash[kafka_key] = self[key]
  end

  kafka_config.symbolize_keys.merge(cfg)
end