Set timeout in produceBatch too

Author: Owen Smith

kafka.c

@@ -468,8 +468,9 @@ int kafka_produce_report(rd_kafka_t *r, const char *topic, char *msg, int msg_le
     return 0;
 }
 
-int kafka_produce_batch(rd_kafka_t *r, char *topic, char **msg, int *msg_len, int msg_cnt, int report)
+int kafka_produce_batch(rd_kafka_t *r, char *topic, char **msg, int *msg_len, int msg_cnt, int report, long timeout)
 {
+    char errstr[512];
     rd_kafka_topic_t *rkt;
     struct produce_cb_params pcb = {msg_cnt, 0, 0, 0, 0, NULL};
     void *opaque;
@@ -496,6 +497,24 @@ int kafka_produce_batch(rd_kafka_t *r, char *topic, char **msg, int *msg_len, in
     /* Topic configuration */
     topic_conf = rd_kafka_topic_conf_new();
 
+    char timeoutStr[64];
+    snprintf(timeoutStr, 64, "%lu", timeout);
+    if (rd_kafka_topic_conf_set(topic_conf, "message.timeout.ms", timeoutStr, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
+    {
+        if (log_level)
+        {
+            openlog("phpkafka", 0, LOG_USER);
+            syslog(
+                LOG_ERR,
+                "Failed to configure topic param 'message.timeout.ms' to %lu before producing; config err was: %s",
+                timeout,
+                errstr
+            );
+        }
+        rd_kafka_topic_conf_destroy(topic_conf);
+        return -3;
+    }
+
     /* Create topic */
     rkt = rd_kafka_topic_new(r, topic, topic_conf);
 

kafka.h

@@ -38,7 +38,7 @@ void kafka_set_log_level(int ll);
 void kafka_set_partition(int partition);
 int kafka_produce(rd_kafka_t *r, char* topic, char* msg, int msg_len, int report, long timeout);
 int kafka_produce_report(rd_kafka_t *r, const char *topic, char *msg, int msg_len, long timeout);
-int kafka_produce_batch(rd_kafka_t *r, char *topic, char **msg, int *msg_len, int msg_cnt, int report);
+int kafka_produce_batch(rd_kafka_t *r, char *topic, char **msg, int *msg_len, int msg_cnt, int report, long timeout);
 rd_kafka_t *kafka_set_connection(rd_kafka_type_t type, const char *b, int report_level, const char *compression);
 rd_kafka_t *kafka_get_connection(kafka_connection_params params, const char *brokers);
 int kafka_consume(rd_kafka_t *r, zval* return_value, char* topic, char* offset, int item_count, int partition);

php_kafka.c

@@ -1140,7 +1140,7 @@ PHP_METHOD(Kafka, produceBatch)
             ++current_idx;
             if (current_idx == 50)
             {
-                status = kafka_produce_batch(connection->producer, topic, msg_batch, msg_batch_len, current_idx, connection->delivery_confirm_mode);
+                status = kafka_produce_batch(connection->producer, topic, msg_batch, msg_batch_len, current_idx, connection->delivery_confirm_mode, timeout);
                 if (status)
                 {
                     if (status < 0)
@@ -1160,7 +1160,7 @@ PHP_METHOD(Kafka, produceBatch)
     }
     if (current_idx)
     {//we still have some messages to produce...
-        status = kafka_produce_batch(connection->producer, topic, msg_batch, msg_batch_len, current_idx, connection->delivery_confirm_mode);
+        status = kafka_produce_batch(connection->producer, topic, msg_batch, msg_batch_len, current_idx, connection->delivery_confirm_mode, timeout);
         if (status)
         {
             if (status < 0)