K2 testing final

examples/incremental_alter_configs.c

@@ -181,7 +181,8 @@ cmd_incremental_alter_configs(rd_kafka_conf_t *conf, int argc, char **argv) {
                 char *config_name     = argv[i * 5 + 3];
                 char *config_value    = argv[i * 5 + 4];
                 rd_kafka_ConfigResource_t *config;
-                rd_kafka_AlterConfigOpType_t op_type;
+                rd_kafka_AlterConfigOpType_t op_type =
+                    RD_KAFKA_ALTER_CONFIG_OP_TYPE_SET;
                 rd_kafka_ResourceType_t restype =
                     !strcmp(restype_s, "TOPIC")    ? RD_KAFKA_RESOURCE_TOPIC
                     : !strcmp(restype_s, "BROKER") ? RD_KAFKA_RESOURCE_BROKER

examples/rdkafka_example.c

@@ -580,7 +580,6 @@ int main(int argc, char **argv) {
                  * Producer
                  */
                 char buf[2048];
-                int sendcnt = 0;
 
                 /* Set up a message delivery report callback.
                  * It will be called once for each message, either on successful
@@ -661,7 +660,6 @@ int main(int argc, char **argv) {
                                         "%s partition %i\n",
                                         len, rd_kafka_topic_name(rkt),
                                         partition);
-                        sendcnt++;
                         /* Poll to handle delivery reports */
                         rd_kafka_poll(rk, 0);
                 }

src/rdbuf.c

@@ -931,6 +931,7 @@ size_t rd_slice_read(rd_slice_t *slice, void *dst, size_t size) {
         }
 
         rd_dassert(remains == 0);
+        (void)remains; /* Only used in asserts */
 
         /* Restore original size */
         slice->end = orig_end;

src/rdkafka.c

@@ -5392,10 +5392,15 @@ rd_kafka_Uuid_t *rd_kafka_Uuid_copy(const rd_kafka_Uuid_t *uuid) {
  */
 rd_kafka_Uuid_t rd_kafka_Uuid_random() {
         int i;
-        unsigned char rand_values_bytes[16] = {0};
-        uint64_t *rand_values_uint64        = (uint64_t *)rand_values_bytes;
+        union {
+                unsigned char bytes[16];
+                uint64_t uint64s[2];
+        } rand_values                    = {{0}};
+        unsigned char *rand_values_bytes = rand_values.bytes;
+        uint64_t *rand_values_uint64     = rand_values.uint64s;
         unsigned char *rand_values_app;
         rd_kafka_Uuid_t ret = RD_KAFKA_UUID_ZERO;
+
         for (i = 0; i < 16; i += 2) {
                 uint16_t rand_uint16 = (uint16_t)rd_jitter(0, INT16_MAX - 1);
                 /* No need to convert endianess here because it's still only

src/rdkafka_broker.c

@@ -897,7 +897,6 @@ static int rd_kafka_broker_bufq_timeout_scan(rd_kafka_broker_t *rkb,
                                              int log_first_n) {
         rd_kafka_buf_t *rkbuf, *tmp;
         int cnt = 0;
-        int idx = -1;
         const rd_kafka_buf_t *holb;
 
 restart:
@@ -906,8 +905,6 @@ static int rd_kafka_broker_bufq_timeout_scan(rd_kafka_broker_t *rkb,
         TAILQ_FOREACH_SAFE(rkbuf, &rkbq->rkbq_bufs, rkbuf_link, tmp) {
                 rd_kafka_broker_state_t pre_state, post_state;
 
-                idx++;
-
                 if (likely(now && rkbuf->rkbuf_ts_timeout > now))
                         continue;
 

src/rdkafka_cgrp.c

@@ -1498,12 +1498,14 @@ static void rd_kafka_cgrp_rejoin(rd_kafka_cgrp_t *rkcg, const char *fmt, ...) {
         char reason[512];
         va_list ap;
         char astr[128];
+
+        va_start(ap, fmt);
+
         if (rkcg->rkcg_group_protocol == RD_KAFKA_GROUP_PROTOCOL_CONSUMER) {
                 rd_kafka_cgrp_consumer_rejoin(rkcg, fmt, ap);
+                va_end(ap);
                 return;
         }
-
-        va_start(ap, fmt);
         rd_vsnprintf(reason, sizeof(reason), fmt, ap);
         va_end(ap);
 
@@ -1677,7 +1679,6 @@ static void rd_kafka_cooperative_protocol_adjust_assignment(
         int i;
         int expected_max_assignment_size;
         int total_assigned = 0;
-        int not_revoking   = 0;
         size_t par_cnt     = 0;
         const rd_kafka_topic_partition_t *toppar;
         const PartitionMemberInfo_t *pmi;
@@ -1743,7 +1744,6 @@ static void rd_kafka_cooperative_protocol_adjust_assignment(
                                                   toppar->partition);
 
                 total_assigned++;
-                not_revoking++;
         }
 
         /* For ready-to-migrate-partitions, it is safe to move them

src/rdkafka_request.c

@@ -3404,10 +3404,10 @@ void rd_kafka_handle_SaslAuthenticate(rd_kafka_t *rk,
                 rd_kafka_buf_read_i64(rkbuf, &session_lifetime_ms);
 
                 if (session_lifetime_ms)
-                        rd_kafka_dbg(
-                            rk, SECURITY, "REAUTH",
-                            "Received session lifetime %ld ms from broker",
-                            session_lifetime_ms);
+                        rd_kafka_dbg(rk, SECURITY, "REAUTH",
+                                     "Received session lifetime %" PRId64
+                                     " ms from broker",
+                                     session_lifetime_ms);
                 rd_kafka_broker_start_reauth_timer(rkb, session_lifetime_ms);
         }
 

src/rdkafka_telemetry_decode.c

@@ -522,17 +522,17 @@ unit_test_telemetry_decode_error(void *opaque, const char *error, ...) {
         rd_assert(!*"Failure while decoding telemetry data");
 }
 
-int unit_test_telemetry(rd_kafka_type_t rk_type,
-                        rd_kafka_telemetry_producer_metric_name_t metric_name,
-                        const char *expected_name,
-                        const char *expected_description,
-                        rd_kafka_telemetry_metric_type_t expected_type,
-                        rd_bool_t is_double,
-                        rd_bool_t is_per_broker,
-                        void (*set_metric_value)(rd_kafka_t *,
-                                                 rd_kafka_broker_t *),
-                        int64_t expected_value_int,
-                        double expected_value_double) {
+int unit_test_telemetry(
+    rd_kafka_type_t rk_type,
+    int metric_name, /* Accepts both producer and consumer metric enums */
+    const char *expected_name,
+    const char *expected_description,
+    rd_kafka_telemetry_metric_type_t expected_type,
+    rd_bool_t is_double,
+    rd_bool_t is_per_broker,
+    void (*set_metric_value)(rd_kafka_t *, rd_kafka_broker_t *),
+    int64_t expected_value_int,
+    double expected_value_double) {
         rd_kafka_t *rk = rd_calloc(1, sizeof(*rk));
         rwlock_init(&rk->rk_lock);
         rd_kafka_conf_t *conf = rd_kafka_conf_new();

src/rdmap.c

@@ -504,5 +504,5 @@ int unittest_map(void) {
         fails += unittest_untyped_map();
         fails += unittest_typed_map();
         fails += unittest_typed_map2();
-        return 0;
+        return fails;
 }

src/snappy.c

@@ -1437,13 +1437,11 @@ static inline int sn_compress(struct snappy_env *env, struct source *reader,
 			   struct sink *writer)
 {
 	int err;
-	size_t written = 0;
 	int N = available(reader);
 	char ulength[kmax32];
 	char *p = varint_encode32(ulength, N);
 
 	append(writer, ulength, p - ulength);
-	written += (p - ulength);
 
 	while (N > 0) {
 		/* Get next block to compress (without copying if possible) */
@@ -1500,7 +1498,6 @@ static inline int sn_compress(struct snappy_env *env, struct source *reader,
 		char *end = compress_fragment(fragment, fragment_size,
 					      dest, table, table_size);
 		append(writer, dest, end - dest);
-		written += (end - dest);
 
 		N -= num_to_read;
 		skip(reader, pending_advance);

tests/0001-multiobj.c

@@ -57,8 +57,11 @@ int main_0001_multiobj(int argc, char **argv) {
 
                 test_conf_init(&conf, &topic_conf, 30);
 
-                if (!topic)
+                if (!topic) {
                         topic = test_mk_topic_name("0001", 0);
+                        test_create_topic_if_auto_create_disabled(NULL, topic,
+                                                                  -1);
+                }
 
                 TIMING_START(&t_full, "full create-produce-destroy cycle");
                 rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
@@ -92,7 +95,7 @@ int main_0001_multiobj(int argc, char **argv) {
 
                 /* Topic is created on the first iteration. */
                 if (i > 0)
-                        TIMING_ASSERT(&t_full, 0, 999);
+                        TIMING_ASSERT(&t_full, 0, tmout_multip(999));
                 else
                         /* Allow metadata propagation. */
                         rd_sleep(1);

tests/0002-unkpart.c

@@ -83,6 +83,7 @@ static void do_test_unkpart(void) {
         int i;
         int fails = 0;
         const struct rd_kafka_metadata *metadata;
+        const char *topic;
 
         TEST_SAY(_C_BLU "%s\n" _C_CLR, __FUNCTION__);
 
@@ -94,7 +95,10 @@ static void do_test_unkpart(void) {
         /* Create kafka instance */
         rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
 
-        rkt = rd_kafka_topic_new(rk, test_mk_topic_name("0002", 0), topic_conf);
+        topic = test_mk_topic_name("0002", 0);
+        test_create_topic_if_auto_create_disabled(rk, topic, 3);
+
+        rkt = rd_kafka_topic_new(rk, topic, topic_conf);
         if (!rkt)
                 TEST_FAIL("Failed to create topic: %s\n",
                           rd_kafka_err2str(rd_kafka_last_error()));
@@ -199,7 +203,9 @@ static void do_test_unkpart_timeout_nobroker(void) {
         rd_kafka_conf_set_dr_msg_cb(conf, test_dr_msg_cb);
         test_curr->exp_dr_err = RD_KAFKA_RESP_ERR__MSG_TIMED_OUT;
 
-        rk  = test_create_handle(RD_KAFKA_PRODUCER, conf);
+        rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
+
+        test_create_topic_if_auto_create_disabled(NULL, topic, 3);
         rkt = rd_kafka_topic_new(rk, topic, NULL);
 
         err = rd_kafka_produce(rkt, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY,

tests/0003-msgmaxsize.c

@@ -79,6 +79,7 @@ int main_0003_msgmaxsize(int argc, char **argv) {
         rd_kafka_conf_t *conf;
         rd_kafka_topic_conf_t *topic_conf;
         char errstr[512];
+        const char *topic;
 
         static const struct {
                 ssize_t keylen;
@@ -108,7 +109,10 @@ int main_0003_msgmaxsize(int argc, char **argv) {
         /* Create kafka instance */
         rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
 
-        rkt = rd_kafka_topic_new(rk, test_mk_topic_name("0003", 0), topic_conf);
+        topic = test_mk_topic_name("0003", 0);
+        test_create_topic_if_auto_create_disabled(NULL, topic, -1);
+        rkt = rd_kafka_topic_new(rk, topic, topic_conf);
+
         if (!rkt)
                 TEST_FAIL("Failed to create topic: %s\n", rd_strerror(errno));