plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
static void
rd_kafka_propagate_consumer_topic_errors (
        rd_kafka_cgrp_t *rkcg, rd_kafka_topic_partition_list_t *errored,
        const char *error_prefix) {
        int i;

        for (i = 0 ; i < errored->cnt ; i++) {
                rd_kafka_topic_partition_t *topic = &errored->elems[i];
                rd_kafka_topic_partition_t *prev;

                rd_assert(topic->err);

                /* Normalize error codes, unknown topic may be
                 * reported by the broker, or the lack of a topic in
                 * metadata response is figured out by the client.
                 * Make sure the application only sees one error code
                 * for both these cases. */
                if (topic->err == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)
                        topic->err = RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART;

                /* Check if this topic errored previously */
                prev = rd_kafka_topic_partition_list_find(
                        rkcg->rkcg_errored_topics, topic->topic,
                        RD_KAFKA_PARTITION_UA);

                if (prev && prev->err == topic->err)
                        continue; /* This topic already reported same error */

                rd_kafka_dbg(rkcg->rkcg_rk, CONSUMER|RD_KAFKA_DBG_TOPIC,
                             "TOPICERR",
                             "%s: %s: %s",
                             error_prefix, topic->topic,
                             rd_kafka_err2str(topic->err));

                /* Send consumer error to application */
                rd_kafka_consumer_err(rkcg->rkcg_q, RD_KAFKA_NODEID_UA,
                                      topic->err, 0,
                                      topic->topic, NULL,
                                      RD_KAFKA_OFFSET_INVALID,
                                      "%s: %s: %s",
                                      error_prefix, topic->topic,
                                      rd_kafka_err2str(topic->err));
        }

        rd_kafka_topic_partition_list_destroy(rkcg->rkcg_errored_topics);
        rkcg->rkcg_errored_topics = errored;
}
/**
 * @brief Generate consumer errors for each topic in the list.
 *
 * Also replaces the list of last reported topic errors so that repeated
 * errors are silenced.
 *
 * @param errored Errored topics.
 * @param error_prefix Error message prefix.
 *
 * @remark Assumes ownership of \p errored.
 */
@brief Generate consumer errors for each topic in the list.
Also replaces the list of last reported topic errors so that repeated
errors are silenced.

@param errored Errored topics.
@param error_prefix Error message prefix.

@remark Assumes ownership of \p errored.
@brief Generate consumer errors for each topic in the list . Also replaces the list of last reported topic errors so that repeated errors are silenced . @param errored Errored topics . @param error_prefix Error message prefix . @remark Assumes ownership of \ p errored .
static void
rd_kafka_propagate_consumer_topic_errors (
        rd_kafka_cgrp_t *rkcg, rd_kafka_topic_partition_list_t *errored,
        const char *error_prefix) {
        int i;
        for (i = 0 ; i < errored->cnt ; i++) {
                rd_kafka_topic_partition_t *topic = &errored->elems[i];
                rd_kafka_topic_partition_t *prev;
                rd_assert(topic->err);
                if (topic->err == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)
                        topic->err = RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART;
                prev = rd_kafka_topic_partition_list_find(
                        rkcg->rkcg_errored_topics, topic->topic,
                        RD_KAFKA_PARTITION_UA);
                if (prev && prev->err == topic->err)
                        continue; 
                rd_kafka_dbg(rkcg->rkcg_rk, CONSUMER|RD_KAFKA_DBG_TOPIC,
                             "TOPICERR",
                             "%s: %s: %s",
                             error_prefix, topic->topic,
                             rd_kafka_err2str(topic->err));
                rd_kafka_consumer_err(rkcg->rkcg_q, RD_KAFKA_NODEID_UA,
                                      topic->err, 0,
                                      topic->topic, NULL,
                                      RD_KAFKA_OFFSET_INVALID,
                                      "%s: %s: %s",
                                      error_prefix, topic->topic,
                                      rd_kafka_err2str(topic->err));
        }
        rd_kafka_topic_partition_list_destroy(rkcg->rkcg_errored_topics);
        rkcg->rkcg_errored_topics = errored;
}
static void rd_kafka_propagate_consumer_topic_errors ( rd_kafka_cgrp_t * rkcg , rd_kafka_topic_partition_list_t * errored , const char * error_prefix ) { int i ; for ( i = 0 ; i < errored -> cnt ; i ++ ) { rd_kafka_topic_partition_t * topic = & errored -> elems [ i ] ; rd_kafka_topic_partition_t * prev ; rd_assert ( topic -> err ) ; if ( topic -> err == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC ) topic -> err = RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART ; prev = rd_kafka_topic_partition_list_find ( rkcg -> rkcg_errored_topics , topic -> topic , RD_KAFKA_PARTITION_UA ) ; if ( prev && prev -> err == topic -> err ) continue ; rd_kafka_dbg ( rkcg -> rkcg_rk , CONSUMER | RD_KAFKA_DBG_TOPIC , " " , " " , error_prefix , topic -> topic , rd_kafka_err2str ( topic -> err ) ) ; rd_kafka_consumer_err ( rkcg -> rkcg_q , RD_KAFKA_NODEID_UA , topic -> err , 0 , topic -> topic , NULL , RD_KAFKA_OFFSET_INVALID , " " , error_prefix , topic -> topic , rd_kafka_err2str ( topic -> err ) ) ; } rd_kafka_topic_partition_list_destroy ( rkcg -> rkcg_errored_topics ) ; rkcg -> rkcg_errored_topics = errored ; }
@brief Generate consumer errors for each topic in the list.
@brief Generate consumer errors for each topic in the list .
/* Normalize error codes, unknown topic may be
                 * reported by the broker, or the lack of a topic in
                 * metadata response is figured out by the client.
                 * Make sure the application only sees one error code
                 * for both these cases. */ /* Check if this topic errored previously */ /* This topic already reported same error */ /* Send consumer error to application */