kafka.rb in semantic_logger-4.12.0

- old
+ new

@@ -22,11 +22,11 @@
 module SemanticLogger
   module Appender
     class Kafka < SemanticLogger::Subscriber
       attr_accessor :seed_brokers, :client_id, :connect_timeout, :socket_timeout,
                     :ssl_ca_cert, :ssl_client_cert, :ssl_client_cert_key, :ssl_ca_certs_from_system,
-                    :delivery_threshold, :delivery_interval,
+                    :delivery_threshold, :delivery_interval, :required_acks,
                     :topic, :partition, :partition_key, :key
 
       # Send log messages to Kafka in JSON format.
       #
       # Kafka Parameters:
@@ -88,10 +88,14 @@
       #
       #    delivery_interval: [Integer]
       #      Number of seconds between triggering a delivery of messages to Apache Kafka.
       #      Default: 5
       #
+      #    required_acks: [Integer]
+      #      Number of replicas that must acknowledge receipt of each log message to the topic
+      #      Default: 1
+      #
       # Semantic Logger Parameters:
       #
       #   level: [:trace | :debug | :info | :warn | :error | :fatal]
       #     Override the log level for this appender.
       #     Default: SemanticLogger.default_level
@@ -119,11 +123,11 @@
       #     Send metrics only events to kafka.
       #     Default: true
       def initialize(seed_brokers:, client_id: "semantic-logger", connect_timeout: nil, socket_timeout: nil,
                      ssl_ca_cert: nil, ssl_client_cert: nil, ssl_client_cert_key: nil, ssl_ca_certs_from_system: false,
                      topic: "log_messages", partition: nil, partition_key: nil, key: nil,
-                     delivery_threshold: 100, delivery_interval: 10,
+                     delivery_threshold: 100, delivery_interval: 10, required_acks: 1,
                      metrics: true, **args, &block)
 
         @seed_brokers             = seed_brokers
         @client_id                = client_id
         @connect_timeout          = connect_timeout
@@ -136,10 +140,11 @@
         @partition                = partition
         @partition_key            = partition_key
         @key                      = key
         @delivery_threshold       = delivery_threshold
         @delivery_interval        = delivery_interval
+        @required_acks            = required_acks
 
         super(metrics: metrics, **args, &block)
         reopen
       end
 
@@ -156,10 +161,11 @@
           logger:                   logger
         )
 
         @producer = @kafka.async_producer(
           delivery_threshold: delivery_threshold,
-          delivery_interval:  delivery_interval
+          delivery_interval:  delivery_interval,
+          required_acks:      required_acks
         )
       end
 
       def close
         @producer&.shutdown