MSVC compatibility about statement

expressions

src/rdkafka.h

@@ -92,6 +92,7 @@ typedef SSIZE_T ssize_t;
 #define RD_DEPRECATED __attribute__((deprecated))
 
 #if defined(__clang__) || defined(__GNUC__) || defined(__GNUG__)
+#define RD_HAS_STATEMENT_EXPRESSIONS
 #define RD_FORMAT(...) __attribute__((format(__VA_ARGS__)))
 #else
 #define RD_FORMAT(...)

src/rdkafka_admin.c

@@ -3576,10 +3576,13 @@ rd_kafka_DeleteRecordsResponse_parse(rd_kafka_op_t *rko_req,
 
         rd_kafka_buf_read_throttle_time(reply);
 
-        offsets = rd_kafka_buf_read_topic_partitions(
-            reply, 0, RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
             RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET,
-            RD_KAFKA_TOPIC_PARTITION_FIELD_ERR);
+            RD_KAFKA_TOPIC_PARTITION_FIELD_ERR,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
+        offsets = rd_kafka_buf_read_topic_partitions(reply, 0, fields);
         if (!offsets)
                 rd_kafka_buf_parse_fail(reply,
                                         "Failed to parse topic partitions");
@@ -4159,9 +4162,12 @@ rd_kafka_OffsetDeleteResponse_parse(rd_kafka_op_t *rko_req,
 
         rd_kafka_buf_read_throttle_time(reply);
 
-        partitions = rd_kafka_buf_read_topic_partitions(
-            reply, 16, RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
-            RD_KAFKA_TOPIC_PARTITION_FIELD_ERR);
+
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_ERR,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
+        partitions = rd_kafka_buf_read_topic_partitions(reply, 16, fields);
         if (!partitions) {
                 rd_snprintf(errstr, errstr_size,
                             "Failed to parse OffsetDeleteResponse partitions");
@@ -6445,8 +6451,11 @@ rd_kafka_DescribeConsumerGroupsResponse_parse(rd_kafka_op_t *rko_req,
                                 /* Decreased in rd_kafka_buf_destroy */
                                 rd_kafka_broker_keep(rkb);
                                 rd_kafka_buf_read_i16(rkbuf, &version);
+                                const rd_kafka_topic_partition_field_t fields[] =
+                                    {RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+                                     RD_KAFKA_TOPIC_PARTITION_FIELD_END};
                                 partitions = rd_kafka_buf_read_topic_partitions(
-                                    rkbuf, 0, rd_false, rd_false);
+                                    rkbuf, 0, fields);
                                 rd_kafka_buf_destroy(rkbuf);
                                 if (!partitions)
                                         rd_kafka_buf_parse_fail(

src/rdkafka_assignor.c

@@ -144,12 +144,15 @@ rd_kafkap_bytes_t *rd_kafka_consumer_protocol_member_metadata_new(
                 /* If there are no owned partitions, this is specified as an
                  * empty array, not NULL. */
                 rd_kafka_buf_write_i32(rkbuf, 0); /* Topic count */
-        else
+        else {
+                const rd_kafka_topic_partition_field_t fields[] = {
+                    RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+                    RD_KAFKA_TOPIC_PARTITION_FIELD_END};
                 rd_kafka_buf_write_topic_partitions(
                     rkbuf, owned_partitions,
                     rd_false /*don't skip invalid offsets*/,
-                    rd_false /*any offset*/,
-                    RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION);
+                    rd_false /*any offset*/, fields);
+        }
 
         /* Get binary buffer and allocate a new Kafka Bytes with a copy. */
         rd_slice_init_full(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf);

src/rdkafka_cgrp.c

@@ -1505,8 +1505,11 @@ static void rd_kafka_cgrp_handle_SyncGroup_memberstate(
                 rkbuf->rkbuf_rkb = rd_kafka_broker_internal(rkcg->rkcg_rk);
 
         rd_kafka_buf_read_i16(rkbuf, &Version);
-        if (!(assignment = rd_kafka_buf_read_topic_partitions(
-                  rkbuf, 0, RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION)))
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
+        if (!(assignment =
+                  rd_kafka_buf_read_topic_partitions(rkbuf, 0, fields)))
                 goto err_parse;
         rd_kafka_buf_read_bytes(rkbuf, &UserData);
 
@@ -1799,9 +1802,12 @@ static int rd_kafka_group_MemberMetadata_consumer_read(
         rd_kafka_buf_read_bytes(rkbuf, &UserData);
         rkgm->rkgm_userdata = rd_kafkap_bytes_copy(&UserData);
 
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
         if (Version >= 1 &&
-            !(rkgm->rkgm_owned = rd_kafka_buf_read_topic_partitions(
-                  rkbuf, 0, RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION)))
+            !(rkgm->rkgm_owned =
+                  rd_kafka_buf_read_topic_partitions(rkbuf, 0, fields)))
                 goto err;
 
         rd_kafka_buf_destroy(rkbuf);

src/rdkafka_partition.h

@@ -102,11 +102,16 @@ rd_kafka_fetch_pos_make(int64_t offset, int32_t leader_epoch) {
         return fetchpos;
 }
 
+#ifdef RD_HAS_STATEMENT_EXPRESSIONS
 #define RD_KAFKA_FETCH_POS(offset, leader_epoch)                               \
         ({                                                                     \
                 rd_kafka_fetch_pos_t _fetchpos = {offset, leader_epoch};       \
                 _fetchpos;                                                     \
         })
+#else
+#define RD_KAFKA_FETCH_POS(offset, leader_epoch)                               \
+        rd_kafka_fetch_pos_make(offset, leader_epoch)
+#endif
 
 
 

src/rdkafka_request.c

@@ -207,7 +207,7 @@ int rd_kafka_err_action(rd_kafka_broker_t *rkb,
  *
  * @returns a newly allocated list on success, or NULL on parse error.
  */
-rd_kafka_topic_partition_list_t *rd_kafka_buf_read_topic_partitions0(
+rd_kafka_topic_partition_list_t *rd_kafka_buf_read_topic_partitions(
     rd_kafka_buf_t *rkbuf,
     size_t estimated_part_cnt,
     const rd_kafka_topic_partition_field_t *fields) {
@@ -305,7 +305,7 @@ rd_kafka_topic_partition_list_t *rd_kafka_buf_read_topic_partitions0(
  *
  * @remark The \p parts list MUST be sorted.
  */
-int rd_kafka_buf_write_topic_partitions0(
+int rd_kafka_buf_write_topic_partitions(
     rd_kafka_buf_t *rkbuf,
     const rd_kafka_topic_partition_list_t *parts,
     rd_bool_t skip_invalid_offsets,
@@ -772,12 +772,14 @@ rd_kafka_resp_err_t rd_kafka_handle_OffsetForLeaderEpoch(
         if (ApiVersion >= 2)
                 rd_kafka_buf_read_throttle_time(rkbuf);
 
-        *offsets = rd_kafka_buf_read_topic_partitions(
-            rkbuf, 0, RD_KAFKA_TOPIC_PARTITION_FIELD_ERR,
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_ERR,
             RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
             ApiVersion >= 1 ? RD_KAFKA_TOPIC_PARTITION_FIELD_EPOCH
                             : RD_KAFKA_TOPIC_PARTITION_FIELD_NOOP,
-            RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET);
+            RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
+        *offsets = rd_kafka_buf_read_topic_partitions(rkbuf, 0, fields);
         if (!*offsets)
                 goto err_parse;
 
@@ -822,14 +824,16 @@ void rd_kafka_OffsetForLeaderEpochRequest(
         rd_kafka_topic_partition_list_sort_by_topic(parts);
 
         /* Write partition list */
-        rd_kafka_buf_write_topic_partitions(
-            rkbuf, parts, rd_false /*include invalid offsets*/,
-            rd_false /*skip valid offsets */,
+        const rd_kafka_topic_partition_field_t fields[] = {
             RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
             /* CurrentLeaderEpoch */
             RD_KAFKA_TOPIC_PARTITION_FIELD_EPOCH,
             /* LeaderEpoch */
-            RD_KAFKA_TOPIC_PARTITION_FIELD_EPOCH);
+            RD_KAFKA_TOPIC_PARTITION_FIELD_EPOCH,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
+        rd_kafka_buf_write_topic_partitions(
+            rkbuf, parts, rd_false /*include invalid offsets*/,
+            rd_false /*skip valid offsets */, fields);
 
         rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0);
 
@@ -1152,10 +1156,12 @@ void rd_kafka_OffsetFetchRequest(rd_kafka_broker_t *rkb,
 
                 /* Write partition list, filtering out partitions with valid
                  * offsets */
+                const rd_kafka_topic_partition_field_t fields[] = {
+                    RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+                    RD_KAFKA_TOPIC_PARTITION_FIELD_END};
                 PartCnt = rd_kafka_buf_write_topic_partitions(
                     rkbuf, parts, rd_false /*include invalid offsets*/,
-                    rd_false /*skip valid offsets */,
-                    RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION);
+                    rd_false /*skip valid offsets */, fields);
         } else {
                 rd_kafka_buf_write_arraycnt_pos(rkbuf);
         }
@@ -1590,10 +1596,13 @@ rd_kafka_OffsetDeleteRequest(rd_kafka_broker_t *rkb,
         /* GroupId */
         rd_kafka_buf_write_str(rkbuf, grpoffsets->group, -1);
 
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
         rd_kafka_buf_write_topic_partitions(
             rkbuf, grpoffsets->partitions,
             rd_false /*dont skip invalid offsets*/, rd_false /*any offset*/,
-            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION);
+            fields);
 
         rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0);
 
@@ -1617,10 +1626,13 @@ rd_kafka_group_MemberState_consumer_write(rd_kafka_buf_t *env_rkbuf,
         rkbuf = rd_kafka_buf_new(1, 100);
         rd_kafka_buf_write_i16(rkbuf, 0); /* Version */
         rd_assert(rkgm->rkgm_assignment);
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
         rd_kafka_buf_write_topic_partitions(
             rkbuf, rkgm->rkgm_assignment,
             rd_false /*don't skip invalid offsets*/, rd_false /* any offset */,
-            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION);
+            fields);
         rd_kafka_buf_write_kbytes(rkbuf, rkgm->rkgm_userdata);
 
         /* Get pointer to binary buffer */
@@ -4006,10 +4018,13 @@ rd_kafka_DeleteRecordsRequest(rd_kafka_broker_t *rkb,
         rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_DeleteRecords, 1,
                                          4 + (partitions->cnt * 100) + 4);
 
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
         rd_kafka_buf_write_topic_partitions(
             rkbuf, partitions, rd_false /*don't skip invalid offsets*/,
-            rd_false /*any offset*/, RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
-            RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET);
+            rd_false /*any offset*/, fields);
 
         /* timeout */
         op_timeout = rd_kafka_confval_get_int(&options->operation_timeout);

src/rdkafka_request.h

@@ -78,33 +78,17 @@ typedef enum {
         RD_KAFKA_TOPIC_PARTITION_FIELD_NOOP,
 } rd_kafka_topic_partition_field_t;
 
-rd_kafka_topic_partition_list_t *rd_kafka_buf_read_topic_partitions0(
+rd_kafka_topic_partition_list_t *rd_kafka_buf_read_topic_partitions(
     rd_kafka_buf_t *rkbuf,
     size_t estimated_part_cnt,
     const rd_kafka_topic_partition_field_t *fields);
 
-#define rd_kafka_buf_read_topic_partitions(rkbuf, estimated_part_cnt, ...)     \
-        ({                                                                     \
-                rd_kafka_topic_partition_field_t _f[] = {                      \
-                    __VA_ARGS__, RD_KAFKA_TOPIC_PARTITION_FIELD_END};          \
-                rd_kafka_buf_read_topic_partitions0(rkbuf, estimated_part_cnt, \
-                                                    _f);                       \
-        })
-
-int rd_kafka_buf_write_topic_partitions0(
+int rd_kafka_buf_write_topic_partitions(
     rd_kafka_buf_t *rkbuf,
     const rd_kafka_topic_partition_list_t *parts,
     rd_bool_t skip_invalid_offsets,
     rd_bool_t only_invalid_offsets,
     const rd_kafka_topic_partition_field_t *fields);
-#define rd_kafka_buf_write_topic_partitions(rkbuf, parts, skip_invalid,        \
-                                            only_invalid, ...)                 \
-        ({                                                                     \
-                rd_kafka_topic_partition_field_t _f[] = {                      \
-                    __VA_ARGS__, RD_KAFKA_TOPIC_PARTITION_FIELD_END};          \
-                rd_kafka_buf_write_topic_partitions0(                          \
-                    rkbuf, parts, skip_invalid, only_invalid, _f);             \
-        })
 
 rd_kafka_resp_err_t
 rd_kafka_FindCoordinatorRequest(rd_kafka_broker_t *rkb,

src/rdkafka_sticky_assignor.c

@@ -1864,9 +1864,12 @@ static rd_kafkap_bytes_t *rd_kafka_sticky_assignor_get_metadata(
 
         rkbuf = rd_kafka_buf_new(1, 100);
         rd_assert(state->prev_assignment != NULL);
-        rd_kafka_buf_write_topic_partitions(
-            rkbuf, state->prev_assignment, rd_false /*skip invalid offsets*/,
-            rd_false /*any offset*/, RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION);
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
+        rd_kafka_buf_write_topic_partitions(rkbuf, state->prev_assignment,
+                                            rd_false /*skip invalid offsets*/,
+                                            rd_false /*any offset*/, fields);
         rd_kafka_buf_write_i32(rkbuf, state->generation_id);
 
         /* Get binary buffer and allocate a new Kafka Bytes with a copy. */

src/rdkafka_txnmgr.c

@@ -1496,9 +1496,11 @@ static void rd_kafka_txn_handle_TxnOffsetCommit(rd_kafka_t *rk,
 
         rd_kafka_buf_read_throttle_time(rkbuf);
 
-        partitions = rd_kafka_buf_read_topic_partitions(
-            rkbuf, 0, RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
-            RD_KAFKA_TOPIC_PARTITION_FIELD_ERR);
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_ERR,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
+        partitions = rd_kafka_buf_read_topic_partitions(rkbuf, 0, fields);
         if (!partitions)
                 goto err_parse;
 
@@ -1705,13 +1707,16 @@ rd_kafka_txn_send_TxnOffsetCommitRequest(rd_kafka_broker_t *rkb,
         }
 
         /* Write per-partition offsets list */
-        cnt = rd_kafka_buf_write_topic_partitions(
-            rkbuf, rko->rko_u.txn.offsets, rd_true /*skip invalid offsets*/,
-            rd_false /*any offset*/, RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
+        const rd_kafka_topic_partition_field_t fields[] = {
+            RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
             RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET,
             ApiVersion >= 2 ? RD_KAFKA_TOPIC_PARTITION_FIELD_EPOCH
                             : RD_KAFKA_TOPIC_PARTITION_FIELD_NOOP,
-            RD_KAFKA_TOPIC_PARTITION_FIELD_METADATA);
+            RD_KAFKA_TOPIC_PARTITION_FIELD_METADATA,
+            RD_KAFKA_TOPIC_PARTITION_FIELD_END};
+        cnt = rd_kafka_buf_write_topic_partitions(
+            rkbuf, rko->rko_u.txn.offsets, rd_true /*skip invalid offsets*/,
+            rd_false /*any offset*/, fields);
         if (!cnt) {
                 /* No valid partition offsets, don't commit. */
                 rd_kafka_buf_destroy(rkbuf);