src/rdkafka_broker.c
static void rd_kafka_broker_timeout_scan (rd_kafka_broker_t *rkb, rd_ts_t now) {
        int inflight_cnt, retry_cnt, outq_cnt;
        int partial_cnt = 0;

	rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));

        /* In-flight requests waiting for response */
        inflight_cnt = rd_kafka_broker_bufq_timeout_scan(
                rkb, 1, &rkb->rkb_waitresps, NULL, -1,
                RD_KAFKA_RESP_ERR__TIMED_OUT, now, "in flight", 5);
	/* Requests in retry queue */
	retry_cnt = rd_kafka_broker_bufq_timeout_scan(
		rkb, 0, &rkb->rkb_retrybufs, NULL, -1,
		RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE, now, "in retry queue", 0);
        /* Requests in local queue not sent yet.
         * partial_cnt is included in outq_cnt and denotes a request
         * that has been partially transmitted. */
        outq_cnt = rd_kafka_broker_bufq_timeout_scan(
                rkb, 0, &rkb->rkb_outbufs, &partial_cnt, -1,
                RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE, now, "in output queue", 0);

        if (inflight_cnt + retry_cnt + outq_cnt + partial_cnt > 0) {
                rd_rkb_log(rkb, LOG_WARNING, "REQTMOUT",
                           "Timed out %i in-flight, %i retry-queued, "
                           "%i out-queue, %i partially-sent requests",
                           inflight_cnt, retry_cnt, outq_cnt, partial_cnt);

                rkb->rkb_req_timeouts += inflight_cnt + outq_cnt;
                rd_atomic64_add(&rkb->rkb_c.req_timeouts,
                                inflight_cnt + outq_cnt);

                /* If this was a partially sent request that timed out, or the
                 * number of timed out requests have reached the
                 * socket.max.fails threshold, we need to take down the
                 * connection. */
                if (partial_cnt > 0 ||
                    (rkb->rkb_rk->rk_conf.socket_max_fails &&
                     rkb->rkb_req_timeouts >=
                     rkb->rkb_rk->rk_conf.socket_max_fails &&
                     rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP)) {
                        char rttinfo[32];
                        /* Print average RTT (if avail) to help diagnose. */
                        rd_avg_calc(&rkb->rkb_avg_rtt, now);
                        if (rkb->rkb_avg_rtt.ra_v.avg)
                                rd_snprintf(rttinfo, sizeof(rttinfo),
                                            " (average rtt %.3fms)",
                                            (float)(rkb->rkb_avg_rtt.ra_v.avg/
                                                    1000.0f));
                        else
                                rttinfo[0] = 0;
                        errno = ETIMEDOUT;
                        rd_kafka_broker_fail(rkb, LOG_ERR,
                                             RD_KAFKA_RESP_ERR__TIMED_OUT,
                                             "%i request(s) timed out: "
                                             "disconnect%s",
                                             rkb->rkb_req_timeouts, rttinfo);
                }
        }
}
/**
 * Scan the wait-response and outbuf queues for message timeouts.
 *
 * Locality: Broker thread
 */
Scan the wait-response and outbuf queues for message timeouts.
Locality: Broker thread
Scan the wait - response and outbuf queues for message timeouts . Locality : Broker thread
static void rd_kafka_broker_timeout_scan (rd_kafka_broker_t *rkb, rd_ts_t now) {
        int inflight_cnt, retry_cnt, outq_cnt;
        int partial_cnt = 0;
	rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
        inflight_cnt = rd_kafka_broker_bufq_timeout_scan(
                rkb, 1, &rkb->rkb_waitresps, NULL, -1,
                RD_KAFKA_RESP_ERR__TIMED_OUT, now, "in flight", 5);
	retry_cnt = rd_kafka_broker_bufq_timeout_scan(
		rkb, 0, &rkb->rkb_retrybufs, NULL, -1,
		RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE, now, "in retry queue", 0);
        outq_cnt = rd_kafka_broker_bufq_timeout_scan(
                rkb, 0, &rkb->rkb_outbufs, &partial_cnt, -1,
                RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE, now, "in output queue", 0);
        if (inflight_cnt + retry_cnt + outq_cnt + partial_cnt > 0) {
                rd_rkb_log(rkb, LOG_WARNING, "REQTMOUT",
                           "Timed out %i in-flight, %i retry-queued, "
                           "%i out-queue, %i partially-sent requests",
                           inflight_cnt, retry_cnt, outq_cnt, partial_cnt);
                rkb->rkb_req_timeouts += inflight_cnt + outq_cnt;
                rd_atomic64_add(&rkb->rkb_c.req_timeouts,
                                inflight_cnt + outq_cnt);
                if (partial_cnt > 0 ||
                    (rkb->rkb_rk->rk_conf.socket_max_fails &&
                     rkb->rkb_req_timeouts >=
                     rkb->rkb_rk->rk_conf.socket_max_fails &&
                     rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP)) {
                        char rttinfo[32];
                        rd_avg_calc(&rkb->rkb_avg_rtt, now);
                        if (rkb->rkb_avg_rtt.ra_v.avg)
                                rd_snprintf(rttinfo, sizeof(rttinfo),
                                            " (average rtt %.3fms)",
                                            (float)(rkb->rkb_avg_rtt.ra_v.avg/
                                                    1000.0f));
                        else
                                rttinfo[0] = 0;
                        errno = ETIMEDOUT;
                        rd_kafka_broker_fail(rkb, LOG_ERR,
                                             RD_KAFKA_RESP_ERR__TIMED_OUT,
                                             "%i request(s) timed out: "
                                             "disconnect%s",
                                             rkb->rkb_req_timeouts, rttinfo);
                }
        }
}
static void rd_kafka_broker_timeout_scan ( rd_kafka_broker_t * rkb , rd_ts_t now ) { int inflight_cnt , retry_cnt , outq_cnt ; int partial_cnt = 0 ; rd_kafka_assert ( rkb -> rkb_rk , thrd_is_current ( rkb -> rkb_thread ) ) ; inflight_cnt = rd_kafka_broker_bufq_timeout_scan ( rkb , 1 , & rkb -> rkb_waitresps , NULL , -1 , RD_KAFKA_RESP_ERR__TIMED_OUT , now , " " , 5 ) ; retry_cnt = rd_kafka_broker_bufq_timeout_scan ( rkb , 0 , & rkb -> rkb_retrybufs , NULL , -1 , RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE , now , " " , 0 ) ; outq_cnt = rd_kafka_broker_bufq_timeout_scan ( rkb , 0 , & rkb -> rkb_outbufs , & partial_cnt , -1 , RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE , now , " " , 0 ) ; if ( inflight_cnt + retry_cnt + outq_cnt + partial_cnt > 0 ) { rd_rkb_log ( rkb , LOG_WARNING , " " , " " " " , inflight_cnt , retry_cnt , outq_cnt , partial_cnt ) ; rkb -> rkb_req_timeouts += inflight_cnt + outq_cnt ; rd_atomic64_add ( & rkb -> rkb_c . req_timeouts , inflight_cnt + outq_cnt ) ; if ( partial_cnt > 0 || ( rkb -> rkb_rk -> rk_conf . socket_max_fails && rkb -> rkb_req_timeouts >= rkb -> rkb_rk -> rk_conf . socket_max_fails && rkb -> rkb_state >= RD_KAFKA_BROKER_STATE_UP ) ) { char rttinfo [ 32 ] ; rd_avg_calc ( & rkb -> rkb_avg_rtt , now ) ; if ( rkb -> rkb_avg_rtt . ra_v . avg ) rd_snprintf ( rttinfo , sizeof ( rttinfo ) , " " , ( float ) ( rkb -> rkb_avg_rtt . ra_v . avg / 1000.0f ) ) ; else rttinfo [ 0 ] = 0 ; errno = ETIMEDOUT ; rd_kafka_broker_fail ( rkb , LOG_ERR , RD_KAFKA_RESP_ERR__TIMED_OUT , " " " " , rkb -> rkb_req_timeouts , rttinfo ) ; } } }
Scan the wait-response and outbuf queues for message timeouts.
Scan the wait - response and outbuf queues for message timeouts .
/* In-flight requests waiting for response */ /* Requests in retry queue */ /* Requests in local queue not sent yet.
         * partial_cnt is included in outq_cnt and denotes a request
         * that has been partially transmitted. */ /* If this was a partially sent request that timed out, or the
                 * number of timed out requests have reached the
                 * socket.max.fails threshold, we need to take down the
                 * connection. */ /* Print average RTT (if avail) to help diagnose. */