Fix compression types read issue in GetTelemetrySubscriptions response with big-endian architectures (#5183)

* Fix compression types read issue in GetTelemetrySubscriptions response for big-endian architectures
* Decrease allocated buffer size in `rd_kafka_PushTelemetryRequest` and explicitly cast the enum

Author: emasab

CHANGELOG.md

@@ -1,3 +1,24 @@
+# librdkafka v2.12.0
+
+librdkafka v2.12.0 is a feature release:
+
+* Fix compression types read issue in GetTelemetrySubscriptions response
+  for big-endian architectures (#5183, @paravoid).
+
+
+## Fixes
+
+### Telemetry fixes
+
+* Issues: #5179 .
+  Fix issue in GetTelemetrySubscriptions with big-endian
+  architectures where wrong values are read as
+  accepted compression types causing the metrics to be sent uncompressed.
+  Happening since 2.5.0. Since 2.10.1 unit tests are failing when run on
+  big-endian architectures (#5183, @paravoid).
+
+
+
 # librdkafka v2.11.1
 
 librdkafka v2.11.1 is a maintenance release:

src/rdkafka_mock_handlers.c

@@ -2696,14 +2696,16 @@ static int rd_kafka_mock_handle_PushTelemetry(rd_kafka_mock_connection_t *mconn,
         rd_kafka_Uuid_t ClientInstanceId;
         int32_t SubscriptionId;
         rd_bool_t terminating;
+        int8_t CompressionType;
         rd_kafka_compression_t compression_type = RD_KAFKA_COMPRESSION_NONE;
         rd_kafkap_bytes_t metrics;
         rd_kafka_resp_err_t err;
 
         rd_kafka_buf_read_uuid(rkbuf, &ClientInstanceId);
         rd_kafka_buf_read_i32(rkbuf, &SubscriptionId);
         rd_kafka_buf_read_bool(rkbuf, &terminating);
-        rd_kafka_buf_read_i8(rkbuf, &compression_type);
+        rd_kafka_buf_read_i8(rkbuf, &CompressionType);
+        compression_type = CompressionType;
         rd_kafka_buf_read_kbytes(rkbuf, &metrics);
 
         void *uncompressed_payload      = NULL;

src/rdkafka_request.c

@@ -6517,15 +6517,14 @@ rd_kafka_PushTelemetryRequest(rd_kafka_broker_t *rkb,
         }
 
         size_t len = sizeof(rd_kafka_Uuid_t) + sizeof(int32_t) +
-                     sizeof(rd_bool_t) + sizeof(compression_type) +
-                     metrics_size;
+                     sizeof(rd_bool_t) + sizeof(int8_t) + metrics_size;
         rkbuf = rd_kafka_buf_new_flexver_request(rkb, RD_KAFKAP_PushTelemetry,
                                                  1, len, rd_true);
 
         rd_kafka_buf_write_uuid(rkbuf, client_instance_id);
         rd_kafka_buf_write_i32(rkbuf, subscription_id);
         rd_kafka_buf_write_bool(rkbuf, terminating);
-        rd_kafka_buf_write_i8(rkbuf, compression_type);
+        rd_kafka_buf_write_i8(rkbuf, (int8_t)compression_type);
 
         rd_dassert(metrics != NULL);
         rd_dassert(metrics_size >= 0);
@@ -6590,10 +6589,12 @@ void rd_kafka_handle_GetTelemetrySubscriptions(rd_kafka_t *rk,
                 rk->rk_telemetry.accepted_compression_types =
                     rd_calloc(arraycnt, sizeof(rd_kafka_compression_t));
 
-                for (i = 0; i < (size_t)arraycnt; i++)
-                        rd_kafka_buf_read_i8(
-                            rkbuf,
-                            &rk->rk_telemetry.accepted_compression_types[i]);
+                for (i = 0; i < (size_t)arraycnt; i++) {
+                        int8_t AcceptedCompressionType;
+                        rd_kafka_buf_read_i8(rkbuf, &AcceptedCompressionType);
+                        rk->rk_telemetry.accepted_compression_types[i] =
+                            AcceptedCompressionType;
+                }
         } else {
                 rk->rk_telemetry.accepted_compression_types_cnt = 1;
                 rk->rk_telemetry.accepted_compression_types =