diff --git a/plugins/in_kafka/in_kafka.c b/plugins/in_kafka/in_kafka.c
index ec1007492b4..df1605af31c 100644
--- a/plugins/in_kafka/in_kafka.c
+++ b/plugins/in_kafka/in_kafka.c
@@ -161,7 +161,7 @@ static int in_kafka_collect(struct flb_input_instance *ins,
     ret = FLB_EVENT_ENCODER_SUCCESS;
 
     while (ret == FLB_EVENT_ENCODER_SUCCESS) {
-        rkm = rd_kafka_consumer_poll(ctx->kafka.rk, 1);
+        rkm = rd_kafka_consumer_poll(ctx->kafka.rk, ctx->poll_timeount_ms);
 
         if (!rkm) {
             break;
@@ -246,6 +246,21 @@ static int in_kafka_init(struct flb_input_instance *ins,
         goto init_error;
     }
 
+    /* Set the kafka poll timeout dependend on wether we run in our own 
+     * or in the main event thread. 
+     * a) run in main event thread: 
+     *    -> minimize the delay we might create
+     * b) run in our own thread: 
+     *    -> optimize for throuput and relay on 'fetch.wait.max.ms' 
+     *       which is set to 500 by default default. lets set it to
+     *       twice that so that increasing fetch.wait.max.ms still
+     *       has an effect.
+     */
+    ctx->poll_timeount_ms = 1;   
+    if(ins->is_threaded) {
+        ctx->poll_timeount_ms = 1000;
+    }
+
     if (ctx->buffer_max_size > 0) {
         ctx->polling_threshold = ctx->buffer_max_size;
 
diff --git a/plugins/in_kafka/in_kafka.h b/plugins/in_kafka/in_kafka.h
index 41328193b0e..7eca5f5341f 100644
--- a/plugins/in_kafka/in_kafka.h
+++ b/plugins/in_kafka/in_kafka.h
@@ -50,6 +50,7 @@ struct flb_in_kafka_config {
     size_t buffer_max_size;          /* Maximum size of chunk allocation */
     size_t polling_threshold;
     bool enable_auto_commit;
+    int poll_timeount_ms;
 };
 
 #endif