plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
static void rd_kafka_cgrp_join_state_serve (rd_kafka_cgrp_t *rkcg) {
        rd_ts_t now = rd_clock();

        if (unlikely(rd_kafka_fatal_error_code(rkcg->rkcg_rk)))
                return;

        switch (rkcg->rkcg_join_state)
        {
        case RD_KAFKA_CGRP_JOIN_STATE_INIT:
                /* If we have a subscription start the join process. */
                if (!rkcg->rkcg_subscription)
                        break;

                if (rd_interval_immediate(&rkcg->rkcg_join_intvl,
					  1000*1000, now) > 0)
                        rd_kafka_cgrp_join(rkcg);
                break;

        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_JOIN:
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_METADATA:
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_SYNC:
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN:
		break;

        case RD_KAFKA_CGRP_JOIN_STATE_ASSIGNED:
	case RD_KAFKA_CGRP_JOIN_STATE_STARTED:
                if (rd_kafka_cgrp_session_timeout_check(rkcg, now))
                        return;
                /* FALLTHRU */
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_REVOKE_REBALANCE_CB:
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_ASSIGN_REBALANCE_CB:
                if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_SUBSCRIPTION &&
                    rd_interval(&rkcg->rkcg_heartbeat_intvl,
                                rkcg->rkcg_rk->rk_conf.
                                group_heartbeat_intvl_ms * 1000, now) > 0)
                        rd_kafka_cgrp_heartbeat(rkcg);
                break;
        }

}
/**
 * Client group's join state handling
 */
Client group's join state handling
Client group ' s join state handling
static void rd_kafka_cgrp_join_state_serve (rd_kafka_cgrp_t *rkcg) {
        rd_ts_t now = rd_clock();
        if (unlikely(rd_kafka_fatal_error_code(rkcg->rkcg_rk)))
                return;
        switch (rkcg->rkcg_join_state)
        {
        case RD_KAFKA_CGRP_JOIN_STATE_INIT:
                if (!rkcg->rkcg_subscription)
                        break;
                if (rd_interval_immediate(&rkcg->rkcg_join_intvl,
					  1000*1000, now) > 0)
                        rd_kafka_cgrp_join(rkcg);
                break;
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_JOIN:
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_METADATA:
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_SYNC:
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN:
		break;
        case RD_KAFKA_CGRP_JOIN_STATE_ASSIGNED:
	case RD_KAFKA_CGRP_JOIN_STATE_STARTED:
                if (rd_kafka_cgrp_session_timeout_check(rkcg, now))
                        return;
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_REVOKE_REBALANCE_CB:
        case RD_KAFKA_CGRP_JOIN_STATE_WAIT_ASSIGN_REBALANCE_CB:
                if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_SUBSCRIPTION &&
                    rd_interval(&rkcg->rkcg_heartbeat_intvl,
                                rkcg->rkcg_rk->rk_conf.
                                group_heartbeat_intvl_ms * 1000, now) > 0)
                        rd_kafka_cgrp_heartbeat(rkcg);
                break;
        }
}
static void rd_kafka_cgrp_join_state_serve ( rd_kafka_cgrp_t * rkcg ) { rd_ts_t now = rd_clock ( ) ; if ( unlikely ( rd_kafka_fatal_error_code ( rkcg -> rkcg_rk ) ) ) return ; switch ( rkcg -> rkcg_join_state ) { case RD_KAFKA_CGRP_JOIN_STATE_INIT : if ( ! rkcg -> rkcg_subscription ) break ; if ( rd_interval_immediate ( & rkcg -> rkcg_join_intvl , 1000 * 1000 , now ) > 0 ) rd_kafka_cgrp_join ( rkcg ) ; break ; case RD_KAFKA_CGRP_JOIN_STATE_WAIT_JOIN : case RD_KAFKA_CGRP_JOIN_STATE_WAIT_METADATA : case RD_KAFKA_CGRP_JOIN_STATE_WAIT_SYNC : case RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN : break ; case RD_KAFKA_CGRP_JOIN_STATE_ASSIGNED : case RD_KAFKA_CGRP_JOIN_STATE_STARTED : if ( rd_kafka_cgrp_session_timeout_check ( rkcg , now ) ) return ; case RD_KAFKA_CGRP_JOIN_STATE_WAIT_REVOKE_REBALANCE_CB : case RD_KAFKA_CGRP_JOIN_STATE_WAIT_ASSIGN_REBALANCE_CB : if ( rkcg -> rkcg_flags & RD_KAFKA_CGRP_F_SUBSCRIPTION && rd_interval ( & rkcg -> rkcg_heartbeat_intvl , rkcg -> rkcg_rk -> rk_conf . group_heartbeat_intvl_ms * 1000 , now ) > 0 ) rd_kafka_cgrp_heartbeat ( rkcg ) ; break ; } }
Client group's join state handling
Client group ' s join state handling
/* If we have a subscription start the join process. */ /* FALLTHRU */