[KIP-932]: Implement api to acknowledge deserialization failure offsets

src/rdkafka.h

@@ -1013,6 +1013,31 @@ RD_EXPORT
 int32_t rd_kafka_topic_partition_get_leader_epoch(
     const rd_kafka_topic_partition_t *rktpar);
 
+/**
+ * @brief Attach an array of offsets to a topic_partition.
+ *
+ * Stores a deep copy of \p offsets on \p rktpar. The offsets array is
+ * currently consumed only by
+ * rd_kafka_share_acknowledge_deserialization_failure().
+ *
+ * Calling this function with \p offsets == NULL or \p offsets_cnt == 0
+ * clears any previously attached offsets without allocating.
+ *
+ * @param rktpar Partition object.
+ * @param offsets Array of offsets. Deep-copied; may be NULL if
+ *                \p offsets_cnt is 0.
+ * @param offsets_cnt Number of entries in \p offsets.
+ *
+ * @remark This function is intended for internal use by client libraries
+ *         built on top of librdkafka to feed
+ *         rd_kafka_share_acknowledge_deserialization_failure(); it is
+ *         not intended for end-user application code.
+ */
+RD_EXPORT
+void rd_kafka_topic_partition_set_offsets(rd_kafka_topic_partition_t *rktpar,
+                                          const int64_t *offsets,
+                                          size_t offsets_cnt);
+
 /**
  * @brief A growable list of Topic+Partitions.
  *
@@ -3419,6 +3444,51 @@ rd_kafka_share_acknowledge_offset(rd_kafka_share_t *rkshare,
                                   int64_t offset,
                                   rd_kafka_share_AcknowledgeType_t type);
 
+/**
+ * @brief Release a batch of offsets that failed client-side deserialization.
+ *
+ * Marks every (topic, partition, offset) tuple in \p partitions as RELEASE
+ * so the broker can redeliver those records. Intended for higher-level client
+ * libraries that perform record deserialization on top of librdkafka and need
+ * to release a batch of records whose payloads could not be deserialized.
+ *
+ * Each entry in \p partitions describes one (topic, partition) and carries
+ * its set of offsets via rd_kafka_topic_partition_set_offsets(). The
+ * \c offset field of each entry is ignored — only the array attached via
+ * rd_kafka_topic_partition_set_offsets() is consumed.
+ *
+ * Intended to be used in implicit acknowledgement mode to release offsets
+ * that hit a deserialization failure.
+ *
+ * Best-effort: every offset is attempted. If some offsets fail (e.g.
+ * partition or offset not currently acquired, or offset is a GAP record),
+ * the remaining offsets are still processed and the returned error
+ * describes the first failure together with aggregate counts of failed
+ * offsets and partitions.
+ *
+ * The caller retains ownership of \p partitions.
+ *
+ * @param rkshare Share consumer instance.
+ * @param partitions Topic-partition list. Each entry's offsets array is
+ *                   set via rd_kafka_topic_partition_set_offsets().
+ *                   Entries without an offsets array attached are skipped.
+ *
+ * @returns NULL on success, or an rd_kafka_error_t* on failure.
+ *          Possible underlying errors include
+ *          RD_KAFKA_RESP_ERR__INVALID_ARG (NULL list) and
+ *          RD_KAFKA_RESP_ERR__STATE (offset not currently acquired, or
+ *          offset is a GAP record). The caller must free the returned
+ *          error with rd_kafka_error_destroy().
+ *
+ * @remark This function is intended for internal use by client libraries
+ *         built on top of librdkafka to release records whose payloads they
+ * could not deserialize; it is not intended for end-user application code.
+ */
+RD_EXPORT
+rd_kafka_error_t *rd_kafka_share_acknowledge_deserialization_failure(
+    rd_kafka_share_t *rkshare,
+    const rd_kafka_topic_partition_list_t *partitions);
+
 /**
  * @brief Asynchronously commit all pending acknowledgements.
  *

src/rdkafka_partition.c

@@ -2920,11 +2920,32 @@ rd_kafka_topic_partition_update(rd_kafka_topic_partition_t *dst,
 
                 dstpriv->topic_id = srcpriv->topic_id;
 
+                /* Deep-copy offsets array if attached. */
+                if (dstpriv->offsets) {
+                        rd_free(dstpriv->offsets);
+                        dstpriv->offsets     = NULL;
+                        dstpriv->offsets_cnt = 0;
+                }
+
+                if (srcpriv->offsets && srcpriv->offsets_cnt > 0) {
+                        dstpriv->offsets     = rd_calloc(srcpriv->offsets_cnt,
+                                                         sizeof(*srcpriv->offsets));
+                        dstpriv->offsets_cnt = srcpriv->offsets_cnt;
+                        memcpy(dstpriv->offsets, srcpriv->offsets,
+                               srcpriv->offsets_cnt *
+                                   sizeof(*srcpriv->offsets));
+                }
+
         } else if ((dstpriv = dst->_private)) {
                 /* No private object in source, reset the fields. */
                 dstpriv->leader_epoch         = -1;
                 dstpriv->current_leader_epoch = -1;
                 dstpriv->topic_id             = RD_KAFKA_UUID_ZERO;
+                if (dstpriv->offsets) {
+                        rd_free(dstpriv->offsets);
+                        dstpriv->offsets     = NULL;
+                        dstpriv->offsets_cnt = 0;
+                }
         }
 }
 
@@ -2963,6 +2984,8 @@ static void rd_kafka_topic_partition_private_destroy(
     rd_kafka_topic_partition_private_t *parpriv) {
         if (parpriv->rktp)
                 rd_kafka_toppar_destroy(parpriv->rktp);
+        if (parpriv->offsets)
+                rd_free(parpriv->offsets);
         rd_free(parpriv);
 }
 
@@ -3059,6 +3082,33 @@ void rd_kafka_topic_partition_set_current_leader_epoch(
         parpriv->current_leader_epoch = current_leader_epoch;
 }
 
+void rd_kafka_topic_partition_set_offsets(rd_kafka_topic_partition_t *rktpar,
+                                          const int64_t *offsets,
+                                          size_t offsets_cnt) {
+        rd_kafka_topic_partition_private_t *parpriv;
+
+        /* Clearing (NULL/0): only allocate the private glue if one already
+         * exists. */
+        if ((offsets == NULL || offsets_cnt == 0) && !rktpar->_private)
+                return;
+
+        parpriv = rd_kafka_topic_partition_get_private(rktpar);
+
+        if (parpriv->offsets) {
+                rd_free(parpriv->offsets);
+                parpriv->offsets     = NULL;
+                parpriv->offsets_cnt = 0;
+        }
+
+        if (offsets && offsets_cnt > 0) {
+                parpriv->offsets     = rd_calloc(offsets_cnt, sizeof(*offsets));
+                parpriv->offsets_cnt = offsets_cnt;
+                memcpy(parpriv->offsets, offsets,
+                       offsets_cnt * sizeof(*offsets));
+        }
+}
+
+
 /**
  * @brief Set offset and leader epoch from a fetchpos.
  */

src/rdkafka_partition.h

@@ -507,6 +507,13 @@ typedef struct rd_kafka_topic_partition_private_s {
         int32_t leader_epoch;
         /** Topic id. */
         rd_kafka_Uuid_t topic_id;
+        /** Optional array of offsets attached via
+         *  rd_kafka_topic_partition_set_offsets(). Currently consumed
+         *  only by rd_kafka_share_acknowledge_deserialization_failure();
+         *  ignored by all other APIs. NULL when unset. */
+        int64_t *offsets;
+        /** Number of entries in \c offsets. Zero when offsets is NULL. */
+        size_t offsets_cnt;
 } rd_kafka_topic_partition_private_t;
 
 

src/rdkafka_share_acknowledgement.c

@@ -1062,6 +1062,119 @@ rd_kafka_share_acknowledge_offset(rd_kafka_share_t *rkshare,
         return err;
 }
 
+rd_kafka_error_t *rd_kafka_share_acknowledge_deserialization_failure(
+    rd_kafka_share_t *rkshare,
+    const rd_kafka_topic_partition_list_t *partitions) {
+        rd_kafka_error_t *error       = NULL;
+        rd_kafka_resp_err_t first_err = RD_KAFKA_RESP_ERR_NO_ERROR;
+        size_t failed_partitions      = 0;
+        size_t failed_offsets         = 0;
+        int i;
+
+        if (unlikely((error = rd_kafka_share_acquire(rkshare)) != NULL))
+                return error;
+
+        if (!partitions) {
+                error = rd_kafka_error_new(RD_KAFKA_RESP_ERR__INVALID_ARG,
+                                           "partitions is NULL");
+                goto done;
+        }
+
+        if (unlikely((first_err = rd_kafka_share_consumer_closed_err(
+                          rkshare)) != RD_KAFKA_RESP_ERR_NO_ERROR)) {
+                error = rd_kafka_error_new(first_err,
+                                           "Share consumer is closed: %s",
+                                           rd_kafka_err2str(first_err));
+                goto done;
+        }
+
+        for (i = 0; i < partitions->cnt; i++) {
+                rd_kafka_topic_partition_t *rktpar = &partitions->elems[i];
+                const rd_kafka_topic_partition_private_t *parpriv;
+                const int64_t *offsets = NULL;
+                size_t offsets_cnt     = 0;
+                size_t j;
+                rd_bool_t partition_failed = rd_false;
+
+                if (!rktpar->topic || rktpar->partition < 0) {
+                        rktpar->err      = RD_KAFKA_RESP_ERR__INVALID_ARG;
+                        partition_failed = rd_true;
+                        if (first_err == RD_KAFKA_RESP_ERR_NO_ERROR)
+                                first_err = rktpar->err;
+                        failed_partitions++;
+                        continue;
+                }
+
+                if ((parpriv = rktpar->_private)) {
+                        offsets     = parpriv->offsets;
+                        offsets_cnt = parpriv->offsets_cnt;
+                }
+                /* No offsets attached: skip silently (no-op). */
+                if (!offsets || offsets_cnt == 0) {
+                        rktpar->err = RD_KAFKA_RESP_ERR_NO_ERROR;
+                        continue;
+                }
+
+                rktpar->err = RD_KAFKA_RESP_ERR_NO_ERROR;
+
+                for (j = 0; j < offsets_cnt; j++) {
+                        int64_t offset = offsets[j];
+                        rd_kafka_share_ack_batch_entry_t *entry;
+                        int64_t idx;
+                        rd_kafka_resp_err_t err;
+
+                        if (offset < 0) {
+                                err = RD_KAFKA_RESP_ERR__INVALID_ARG;
+                                goto record_err;
+                        }
+
+                        err = rd_kafka_share_find_ack_entry(
+                            rkshare, rktpar->topic, rktpar->partition, offset,
+                            &entry, &idx);
+                        if (err)
+                                goto record_err;
+
+                        /* GAP records cannot be released. */
+                        if (entry->types[idx] ==
+                            RD_KAFKA_SHARE_INTERNAL_ACK_GAP) {
+                                err = RD_KAFKA_RESP_ERR__STATE;
+                                goto record_err;
+                        }
+
+                        rd_kafka_share_update_acknowledgement_type(
+                            rkshare, entry, idx,
+                            RD_KAFKA_SHARE_ACKNOWLEDGE_TYPE_RELEASE);
+                        continue;
+
+                record_err:
+                        failed_offsets++;
+                        partition_failed = rd_true;
+                        /* Record the partition-level err with the first
+                         * per-offset failure for this partition. */
+                        if (rktpar->err == RD_KAFKA_RESP_ERR_NO_ERROR)
+                                rktpar->err = err;
+                        if (first_err == RD_KAFKA_RESP_ERR_NO_ERROR)
+                                first_err = err;
+                }
+
+                if (partition_failed)
+                        failed_partitions++;
+        }
+
+        if (first_err != RD_KAFKA_RESP_ERR_NO_ERROR)
+                error = rd_kafka_error_new(first_err,
+                                           "Failed to release %" PRIusz
+                                           " offset(s) across "
+                                           "%" PRIusz
+                                           " partition(s) (first error: %s)",
+                                           failed_offsets, failed_partitions,
+                                           rd_kafka_err2str(first_err));
+
+done:
+        rd_kafka_share_release(rkshare);
+        return error;
+}
+
 
 /**
  * @brief Initialize a partition offsets element with topicid,topic, partition,