Callbacks now work on bad config. (KxSystems#102)

Also removed invalid fetch.wait.max.ms for producers
sshanks-kx · Sep 8, 2023 · e3d8a39 · e3d8a39
1 parent 5af24b7
commit e3d8a39
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 6 deletions.
diff --git a/examples/test_offsetp.q b/examples/test_offsetp.q
@@ -2,8 +2,7 @@
 kfk_cfg:(!) . flip(
   (`metadata.broker.list;`localhost:9092);
   (`statistics.interval.ms;`10000);
-  (`queue.buffering.max.ms;`1);
-  (`fetch.wait.max.ms;`10)
+  (`queue.buffering.max.ms;`1)
   );
 producer:.kfk.Producer[kfk_cfg]
 

diff --git a/examples/test_producer.q b/examples/test_producer.q
@@ -2,8 +2,7 @@
 kfk_cfg:(!) . flip(
   (`metadata.broker.list;`localhost:9092);
   (`statistics.interval.ms;`10000);
-  (`queue.buffering.max.ms;`1);
-  (`fetch.wait.max.ms;`10)
+  (`queue.buffering.max.ms;`1)
   );
 producer:.kfk.Producer[kfk_cfg]
 

diff --git a/kfk.c b/kfk.c
@@ -223,15 +223,15 @@ EXP K2(kfkClient){
     return krr((S) b);
   if(!(rk= rd_kafka_new(type, conf, b, sizeof(b))))
     return krr(b);
-  /* Redirect logs to main queue */
-  rd_kafka_set_log_queue(rk,NULL);
   /* Redirect rd_kafka_poll() to consumer_poll() */
   if(type == RD_KAFKA_CONSUMER){
     rd_kafka_poll_set_consumer(rk);
     rd_kafka_queue_io_event_enable(rd_kafka_queue_get_consumer(rk),spair[1],"X",1);
   }
   else
     rd_kafka_queue_io_event_enable(rd_kafka_queue_get_main(rk),spair[1],"X",1);
+  /* Redirect logs to main queue */
+  rd_kafka_set_log_queue(rk,NULL);
   js(&clients, (S) rk);
   return ki(clients->n - 1);
 }