PR feedback

confluentinc · Jun 10, 2024 · 4c0d964 · 4c0d964
1 parent 396f7cc
commit 4c0d964
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 12 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+# vNext
+
+* Fix segfault when using long client id because of erased segment when using flexver. (#4689)
+
 # librdkafka v2.4.0
 
 librdkafka v2.4.0 is a feature release:

diff --git a/src/rdbuf.c b/src/rdbuf.c
@@ -667,8 +667,8 @@ size_t rd_buf_erase(rd_buf_t *rbuf, size_t absof, size_t size) {
 
                 /* If segment is now empty, remove it */
                 if (seg->seg_of == 0) {
+                        rbuf->rbuf_erased -= seg->seg_erased;
                         rd_buf_destroy_segment(rbuf, seg);
-                        rbuf->rbuf_erased -= toerase;
                 }
         }
 
@@ -712,8 +712,8 @@ int rd_buf_write_seek(rd_buf_t *rbuf, size_t absof) {
              next != seg;) {
                 rd_segment_t *this = next;
                 next = TAILQ_PREV(this, rd_segment_head, seg_link);
+                rbuf->rbuf_erased -= this->seg_erased;
                 rd_buf_destroy_segment(rbuf, this);
-                rbuf->rbuf_erased -= seg->seg_erased;
         }
 
         /* Update relative write offset */

diff --git a/tests/0011-produce_batch.c b/tests/0011-produce_batch.c
@@ -90,20 +90,18 @@ static void test_single_partition(void) {
         int failcnt = 0;
         int i;
         rd_kafka_message_t *rkmessages;
-        const int client_id_length = 239;
-        char *client_id;
-
+        char client_id[271];
         SUB_TEST_QUICK();
 
         msgid_next = 0;
 
         test_conf_init(&conf, &topic_conf, 20);
 
-        client_id = malloc((client_id_length + 1) * sizeof(char));
-        for (i = 0; i < client_id_length; i++) {
-                client_id[i] = 'c';
-        }
-        client_id[client_id_length] = '\0';
+        /* A long client id must not cause a segmentation fault
+         * because of an erased segment when using flexver.
+         * See: https://github.com/confluentinc/confluent-kafka-dotnet/issues/2084 */
+        memset(client_id, 'c', sizeof(client_id) - 1);
+        client_id[sizeof(client_id) - 1] = '\0';
         rd_kafka_conf_set(conf, "client.id", client_id, NULL, 0);
 
         /* Set delivery report callback */
@@ -187,8 +185,6 @@ static void test_single_partition(void) {
         TEST_SAY("Destroying kafka instance %s\n", rd_kafka_name(rk));
         rd_kafka_destroy(rk);
 
-        free(client_id);
-
         SUB_TEST_PASS();
 }