commit 495747599cebdc141e3e1d6f960a9e22da06fd6d
Author: John Skopis <jskopis@cloudflare.com>
Date:   Tue Jan 14 12:55:28 2020 +0000

    [BUG] Fix kafka deadlock
    
    rdkafka internally maintains a message buffer. syslog-ng fills the
    rdkafka buffer optimistically. When the rdkafka buffer becomes full,
    typically the result of a kafka failure, or there is a kafka failure the
    kafka dest fifo becomes full. When the kafka dest fifo is full messages
    are dropped.
    
    Seems the worker thread is only scheduled when there is work to do, i.e.
    messages are inserted into the fifo. When the fifo is full messages are
    never inserted.
    
    When the worker thread is not scheduled it seems the periodic,
    `_drain_responses`, is also not scheduled so `rd_kafka_poll` is never
    called and the internal rdkafka buffer is full of delivery reports that
    need to be handled.
    
    The simple fix here is to move the periodic out of the worker and onto
    the driver. I don't think `rd_kafka_poll` is very expensive so this
    approach is probably fine. A better improvement would be to introduce a
    core module solely to run periodics within a target deadline.

diff --git modules/kafka/kafka-dest-driver.c modules/kafka/kafka-dest-driver.c
index 301dff104..5596a4b1f 100644
--- modules/kafka/kafka-dest-driver.c
+++ modules/kafka/kafka-dest-driver.c
@@ -27,6 +27,7 @@
 #include "kafka-dest-driver.h"
 #include "kafka-props.h"
 #include "kafka-dest-worker.h"
+#include "timeutils/misc.h"
 
 #include <librdkafka/rdkafka.h>
 #include <stdlib.h>
@@ -324,6 +325,34 @@ _purge_remaining_messages(KafkaDestDriver *self)
 
 }
 
+static void
+_update_drain_timer(KafkaDestDriver *self)
+{
+  if (iv_timer_registered(&self->poll_timer))
+    iv_timer_unregister(&self->poll_timer);
+  iv_validate_now();
+  self->poll_timer.expires = iv_now;
+  timespec_add_msec(&self->poll_timer.expires, self->poll_timeout);
+  iv_timer_register(&self->poll_timer);
+  msg_trace("kafka: timer rescheduled");
+}
+
+static void
+_drain_responses(KafkaDestDriver *self)
+{
+  _update_drain_timer(self);
+  gint count = rd_kafka_poll(self->kafka, 0);
+  if (count != 0)
+    {
+      msg_trace("kafka: destination side rd_kafka_poll() processed some responses",
+                evt_tag_str("topic", self->topic_name),
+                evt_tag_int("count", count),
+                evt_tag_str("driver", self->super.super.super.id),
+                log_pipe_location_tag(&self->super.super.super.super));
+    }
+}
+
+
 static gboolean
 kafka_dd_init(LogPipe *s)
 {
@@ -375,6 +404,14 @@ kafka_dd_init(LogPipe *s)
               evt_tag_str("driver", self->super.super.super.id),
               log_pipe_location_tag(&self->super.super.super.super));
 
+  IV_TIMER_INIT(&self->poll_timer);
+  self->poll_timer.cookie = self;
+  self->poll_timer.handler = (void (*)(void *)) _drain_responses;
+  self->poll_timer.expires = iv_now;
+  self->poll_timer.expires.tv_sec++;
+  self->poll_timer.expires.tv_nsec = 0;
+  iv_timer_register(&self->poll_timer);
+
   return log_threaded_dest_driver_start_workers(&self->super);
 }
 
diff --git modules/kafka/kafka-dest-driver.h modules/kafka/kafka-dest-driver.h
index 9dab4758b..4a09a428e 100644
--- modules/kafka/kafka-dest-driver.h
+++ modules/kafka/kafka-dest-driver.h
@@ -44,6 +44,7 @@ typedef struct
   gint flush_timeout_on_shutdown;
   gint flush_timeout_on_reload;
   gint poll_timeout;
+  struct iv_timer poll_timer;
 } KafkaDestDriver;
 
 void kafka_dd_set_topic(LogDriver *d, const gchar *topic);
diff --git modules/kafka/kafka-dest-worker.c modules/kafka/kafka-dest-worker.c
index 903a13fce..119421dd4 100644
--- modules/kafka/kafka-dest-worker.c
+++ modules/kafka/kafka-dest-worker.c
@@ -25,14 +25,12 @@
 #include "kafka-dest-worker.h"
 #include "kafka-dest-driver.h"
 #include "str-utils.h"
-#include "timeutils/misc.h"
 #include <zlib.h>
 
 
 typedef struct _KafkaDestWorker
 {
   LogThreadedDestWorker super;
-  struct iv_timer poll_timer;
   GString *key;
   GString *message;
 } KafkaDestWorker;
@@ -83,45 +81,6 @@ _publish_message(KafkaDestWorker *self, LogMessage *msg)
   return TRUE;
 }
 
-static void
-_update_drain_timer(KafkaDestWorker *self)
-{
-  KafkaDestDriver *owner = (KafkaDestDriver *) self->super.owner;
-
-  if (self->super.worker_index != 0)
-    return;
-
-  if (iv_timer_registered(&self->poll_timer))
-    iv_timer_unregister(&self->poll_timer);
-  iv_validate_now();
-  self->poll_timer.expires = iv_now;
-  timespec_add_msec(&self->poll_timer.expires, owner->poll_timeout);
-  iv_timer_register(&self->poll_timer);
-}
-
-static void
-_drain_responses(KafkaDestWorker *self)
-{
-  KafkaDestDriver *owner = (KafkaDestDriver *) self->super.owner;
-
-  /* we are only draining responses in the first worker thread, so all
-   * callbacks originate from the same thread and we don't need to
-   * synchronize between workers */
-  if (self->super.worker_index != 0)
-    return;
-
-  gint count = rd_kafka_poll(owner->kafka, 0);
-  if (count != 0)
-    {
-      msg_trace("kafka: destination side rd_kafka_poll() processed some responses",
-                evt_tag_str("topic", owner->topic_name),
-                evt_tag_int("count", count),
-                evt_tag_str("driver", owner->super.super.super.id),
-                log_pipe_location_tag(&owner->super.super.super.super));
-    }
-  _update_drain_timer(self);
-}
-
 /*
  * Worker thread
  */
@@ -131,13 +90,10 @@ kafka_dest_worker_insert(LogThreadedDestWorker *s, LogMessage *msg)
 {
   KafkaDestWorker *self = (KafkaDestWorker *)s;
 
-  _drain_responses(self);
-
   _format_message_and_key(self, msg);
   if (!_publish_message(self, msg))
     return LTR_RETRY;
 
-  _drain_responses(self);
   return LTR_SUCCESS;
 }
 
@@ -159,10 +115,6 @@ kafka_dest_worker_new(LogThreadedDestDriver *o, gint worker_index)
   self->super.insert = kafka_dest_worker_insert;
   self->super.free_fn = kafka_dest_worker_free;
 
-  IV_TIMER_INIT(&self->poll_timer);
-  self->poll_timer.cookie = self;
-  self->poll_timer.handler = (void (*)(void *)) _drain_responses;
-
   self->key = g_string_sized_new(0);
   self->message = g_string_sized_new(1024);
   return &self->super;
