src/rdkafka.c
int rd_kafka_set_fatal_error (rd_kafka_t *rk, rd_kafka_resp_err_t err,
                              const char *fmt, ...) {
        va_list ap;
        char buf[512];

        rd_kafka_wrlock(rk);
        rk->rk_fatal.cnt++;
        if (rd_atomic32_get(&rk->rk_fatal.err)) {
                rd_kafka_wrunlock(rk);
                rd_kafka_dbg(rk, GENERIC, "FATAL",
                             "Suppressing subsequent fatal error: %s",
                             rd_kafka_err2name(err));
                return 0;
        }

        rd_atomic32_set(&rk->rk_fatal.err, err);

        va_start(ap, fmt);
        rd_vsnprintf(buf, sizeof(buf), fmt, ap);
        va_end(ap);
        rk->rk_fatal.errstr = rd_strdup(buf);

        rd_kafka_wrunlock(rk);

        /* If there is an error callback or event handler we
         * also log the fatal error as it happens.
         * If there is no error callback the error event
         * will be automatically logged, and this check here
         * prevents us from duplicate logs. */
        if (rk->rk_conf.enabled_events & RD_KAFKA_EVENT_ERROR)
                rd_kafka_log(rk, LOG_EMERG, "FATAL",
                             "Fatal error: %s: %s",
                             rd_kafka_err2str(err), rk->rk_fatal.errstr);
        else
                rd_kafka_dbg(rk, ALL, "FATAL",
                             "Fatal error: %s: %s",
                             rd_kafka_err2str(err), rk->rk_fatal.errstr);

        /* Indicate to the application that a fatal error was raised,
         * the app should use rd_kafka_fatal_error() to extract the
         * fatal error code itself. */
        rd_kafka_op_err(rk, RD_KAFKA_RESP_ERR__FATAL,
                        "Fatal error: %s: %s",
                        rd_kafka_err2str(err), rk->rk_fatal.errstr);


        /* Purge producer queues, but not in-flight since we'll
         * want proper delivery status for transmitted requests.
         * Need NON_BLOCKING to avoid dead-lock if user is
         * calling purge() at the same time, which could be
         * waiting for this broker thread to handle its
         * OP_PURGE request. */
        if (rk->rk_type == RD_KAFKA_PRODUCER)
                rd_kafka_purge(rk, RD_KAFKA_PURGE_F_QUEUE|
                               RD_KAFKA_PURGE_F_NON_BLOCKING);

        return 1;
}
/**
 * @brief Set's the fatal error for this instance.
 *
 * @returns 1 if the error was set, or 0 if a previous fatal error
 *          has already been set on this instance.
 *
 * @locality any
 * @locks none
 */
@brief Set's the fatal error for this instance.
@returns 1 if the error was set, or 0 if a previous fatal error
has already been set on this instance.

@locality any
@locks none
@brief Set ' s the fatal error for this instance . @returns 1 if the error was set or 0 if a previous fatal error has already been set on this instance . @locality any @locks none
int rd_kafka_set_fatal_error (rd_kafka_t *rk, rd_kafka_resp_err_t err,
                              const char *fmt, ...) {
        va_list ap;
        char buf[512];
        rd_kafka_wrlock(rk);
        rk->rk_fatal.cnt++;
        if (rd_atomic32_get(&rk->rk_fatal.err)) {
                rd_kafka_wrunlock(rk);
                rd_kafka_dbg(rk, GENERIC, "FATAL",
                             "Suppressing subsequent fatal error: %s",
                             rd_kafka_err2name(err));
                return 0;
        }
        rd_atomic32_set(&rk->rk_fatal.err, err);
        va_start(ap, fmt);
        rd_vsnprintf(buf, sizeof(buf), fmt, ap);
        va_end(ap);
        rk->rk_fatal.errstr = rd_strdup(buf);
        rd_kafka_wrunlock(rk);
        if (rk->rk_conf.enabled_events & RD_KAFKA_EVENT_ERROR)
                rd_kafka_log(rk, LOG_EMERG, "FATAL",
                             "Fatal error: %s: %s",
                             rd_kafka_err2str(err), rk->rk_fatal.errstr);
        else
                rd_kafka_dbg(rk, ALL, "FATAL",
                             "Fatal error: %s: %s",
                             rd_kafka_err2str(err), rk->rk_fatal.errstr);
        rd_kafka_op_err(rk, RD_KAFKA_RESP_ERR__FATAL,
                        "Fatal error: %s: %s",
                        rd_kafka_err2str(err), rk->rk_fatal.errstr);
        if (rk->rk_type == RD_KAFKA_PRODUCER)
                rd_kafka_purge(rk, RD_KAFKA_PURGE_F_QUEUE|
                               RD_KAFKA_PURGE_F_NON_BLOCKING);
        return 1;
}
int rd_kafka_set_fatal_error ( rd_kafka_t * rk , rd_kafka_resp_err_t err , const char * fmt , ... ) { va_list ap ; char buf [ 512 ] ; rd_kafka_wrlock ( rk ) ; rk -> rk_fatal . cnt ++ ; if ( rd_atomic32_get ( & rk -> rk_fatal . err ) ) { rd_kafka_wrunlock ( rk ) ; rd_kafka_dbg ( rk , GENERIC , " " , " " , rd_kafka_err2name ( err ) ) ; return 0 ; } rd_atomic32_set ( & rk -> rk_fatal . err , err ) ; va_start ( ap , fmt ) ; rd_vsnprintf ( buf , sizeof ( buf ) , fmt , ap ) ; va_end ( ap ) ; rk -> rk_fatal . errstr = rd_strdup ( buf ) ; rd_kafka_wrunlock ( rk ) ; if ( rk -> rk_conf . enabled_events & RD_KAFKA_EVENT_ERROR ) rd_kafka_log ( rk , LOG_EMERG , " " , " " , rd_kafka_err2str ( err ) , rk -> rk_fatal . errstr ) ; else rd_kafka_dbg ( rk , ALL , " " , " " , rd_kafka_err2str ( err ) , rk -> rk_fatal . errstr ) ; rd_kafka_op_err ( rk , RD_KAFKA_RESP_ERR__FATAL , " " , rd_kafka_err2str ( err ) , rk -> rk_fatal . errstr ) ; if ( rk -> rk_type == RD_KAFKA_PRODUCER ) rd_kafka_purge ( rk , RD_KAFKA_PURGE_F_QUEUE | RD_KAFKA_PURGE_F_NON_BLOCKING ) ; return 1 ; }
@brief Set's the fatal error for this instance.
@brief Set ' s the fatal error for this instance .
/* If there is an error callback or event handler we
         * also log the fatal error as it happens.
         * If there is no error callback the error event
         * will be automatically logged, and this check here
         * prevents us from duplicate logs. */ /* Indicate to the application that a fatal error was raised,
         * the app should use rd_kafka_fatal_error() to extract the
         * fatal error code itself. */ /* Purge producer queues, but not in-flight since we'll
         * want proper delivery status for transmitted requests.
         * Need NON_BLOCKING to avoid dead-lock if user is
         * calling purge() at the same time, which could be
         * waiting for this broker thread to handle its
         * OP_PURGE request. */