Revert setting timeout to infinity (confluentinc#5201)

* Revert setting timeout to infinity

* style fix

* Changelog change

* Changelog changes

* Changelog change

Author: PratRanj07

CHANGELOG.md

@@ -9,6 +9,8 @@ librdkafka v2.12.0 is a feature release:
 * Fix for an error being raised after a commit due to an existing error in the topic partition (#4672).
 * Additional KIP-320 related validation: when assigning the offset is validated
   if leader epoch is specified (#4931).
+* Fix double free of headers in `rd_kafka_produceva` method (@blindspotbounty, #4628).
+* Fix to ensure `rd_kafka_query_watermark_offsets` enforces the specified timeout and does not continue beyond timeout expiry (#5201).
 
 
 ## Fixes
@@ -23,6 +25,9 @@ librdkafka v2.12.0 is a feature release:
   so it's possible it did overlap with the periodic `topic.metadata.refresh.interval.ms`
   and cause a re-bootstrap even if not needed.
   Happening since 2.11.0 (#5177).
+* Issues: #4878.
+  Fix to ensure `rd_kafka_query_watermark_offsets` enforces the specified timeout and does not continue beyond timeout expiry.
+  Happening since 2.3.0 (#5201).
 
 ### Telemetry fixes
 
@@ -45,6 +50,12 @@ librdkafka v2.12.0 is a feature release:
   from last available offset of given leader epoch.
   Happens since 2.1.0 (#4931).
 
+### Producer fixes
+
+* Issues: #4627.
+  Fix double free of headers in `rd_kafka_produceva` method in cases where the partition doesn't exist.
+  Happening since 1.x (@blindspotbounty, #4628).
+
 
 
 # librdkafka v2.11.1

src/rdkafka.c

@@ -3805,6 +3805,7 @@ rd_kafka_resp_err_t rd_kafka_query_watermark_offsets(rd_kafka_t *rk,
         struct rd_kafka_partition_leader *leader;
         rd_list_t leaders;
         rd_kafka_resp_err_t err;
+        int tmout;
 
         partitions = rd_kafka_topic_partition_list_new(1);
         rktpar =
@@ -3851,9 +3852,13 @@ rd_kafka_resp_err_t rd_kafka_query_watermark_offsets(rd_kafka_t *rk,
 
         /* Wait for reply (or timeout) */
         while (state.err == RD_KAFKA_RESP_ERR__IN_PROGRESS) {
-                rd_kafka_q_serve(rkq, RD_POLL_INFINITE, 0,
-                                 RD_KAFKA_Q_CB_CALLBACK, rd_kafka_poll_cb,
-                                 NULL);
+                tmout = rd_timeout_remains(ts_end);
+                if (rd_timeout_expired(tmout)) {
+                        state.err = RD_KAFKA_RESP_ERR__TIMED_OUT;
+                        break;
+                }
+                rd_kafka_q_serve(rkq, tmout, 0, RD_KAFKA_Q_CB_CALLBACK,
+                                 rd_kafka_poll_cb, NULL);
         }
 
         rd_kafka_q_destroy_owner(rkq);