From d82bb59f23bbda5a5821ffc18a94cb4fce47d1f3 Mon Sep 17 00:00:00 2001 From: PratRanj07 Date: Mon, 29 Jun 2026 12:19:08 +0530 Subject: [PATCH 1/2] add logging for ack apis --- src/rdkafka_share_acknowledgement.c | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/src/rdkafka_share_acknowledgement.c b/src/rdkafka_share_acknowledgement.c index 35b50d3a57..d81a22cc26 100644 --- a/src/rdkafka_share_acknowledgement.c +++ b/src/rdkafka_share_acknowledgement.c @@ -978,8 +978,13 @@ rd_kafka_share_acknowledge_offset0(rd_kafka_share_t *rkshare, return RD_KAFKA_RESP_ERR__INVALID_ARG; /* Explicit acknowledge APIs require explicit acknowledgement mode */ - if (rd_kafka_share_acknowledgement_mode_is_implicit(rkshare)) + if (rd_kafka_share_acknowledgement_mode_is_implicit(rkshare)) { + rd_kafka_dbg(rkshare->rkshare_rk, CONSUMER, "SHAREACK", + "acknowledge(%s [%" PRId32 "] @%" PRId64 + ") rejected: implicit acknowledgement mode", + topic, partition, offset); return RD_KAFKA_RESP_ERR__STATE; + } /* Validate type - ACCEPT, RELEASE, REJECT allowed */ if (type < RD_KAFKA_SHARE_ACKNOWLEDGE_TYPE_ACCEPT || @@ -989,12 +994,23 @@ rd_kafka_share_acknowledge_offset0(rd_kafka_share_t *rkshare, /* Find partition and entry containing the offset */ err = rd_kafka_share_find_ack_entry(rkshare, topic, partition, offset, &entry, &idx); - if (err) + if (err) { + rd_kafka_dbg(rkshare->rkshare_rk, CONSUMER, "SHAREACK", + "acknowledge(%s [%" PRId32 "] @%" PRId64 + ") rejected: offset not in any in-flight " + "acquired batch", + topic, partition, offset); return err; + } /* GAP records cannot be acknowledged */ - if (entry->types[idx] == RD_KAFKA_SHARE_INTERNAL_ACK_GAP) + if (entry->types[idx] == RD_KAFKA_SHARE_INTERNAL_ACK_GAP) { + rd_kafka_dbg(rkshare->rkshare_rk, CONSUMER, "SHAREACK", + "acknowledge(%s [%" PRId32 "] @%" PRId64 + ") rejected: offset is a GAP record", + topic, partition, offset); return RD_KAFKA_RESP_ERR__STATE; + } rd_kafka_share_update_acknowledgement_type(rkshare, entry, idx, type); From 3ea6576737f25ab640be0dfb75e3154db9c26f44 Mon Sep 17 00:00:00 2001 From: Pratyush Ranjan Date: Thu, 30 Jul 2026 12:16:30 +0530 Subject: [PATCH 2/2] Check --- src/rdkafka_share_acknowledgement.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/rdkafka_share_acknowledgement.c b/src/rdkafka_share_acknowledgement.c index d81a22cc26..5c14937df1 100644 --- a/src/rdkafka_share_acknowledgement.c +++ b/src/rdkafka_share_acknowledgement.c @@ -996,7 +996,7 @@ rd_kafka_share_acknowledge_offset0(rd_kafka_share_t *rkshare, &entry, &idx); if (err) { rd_kafka_dbg(rkshare->rkshare_rk, CONSUMER, "SHAREACK", - "acknowledge(%s [%" PRId32 "] @%" PRId64 + "Acknowledge(%s [%" PRId32 "] @%" PRId64 ") rejected: offset not in any in-flight " "acquired batch", topic, partition, offset); @@ -1006,7 +1006,7 @@ rd_kafka_share_acknowledge_offset0(rd_kafka_share_t *rkshare, /* GAP records cannot be acknowledged */ if (entry->types[idx] == RD_KAFKA_SHARE_INTERNAL_ACK_GAP) { rd_kafka_dbg(rkshare->rkshare_rk, CONSUMER, "SHAREACK", - "acknowledge(%s [%" PRId32 "] @%" PRId64 + "Acknowledge(%s [%" PRId32 "] @%" PRId64 ") rejected: offset is a GAP record", topic, partition, offset); return RD_KAFKA_RESP_ERR__STATE;