src/rdkafka_request.c
void rd_kafka_handle_SyncGroup (rd_kafka_t *rk,
				rd_kafka_broker_t *rkb,
                                rd_kafka_resp_err_t err,
                                rd_kafka_buf_t *rkbuf,
                                rd_kafka_buf_t *request,
                                void *opaque) {
        rd_kafka_cgrp_t *rkcg = opaque;
        const int log_decode_errors = LOG_ERR;
        int16_t ErrorCode = 0;
        rd_kafkap_bytes_t MemberState = RD_ZERO_INIT;
        int actions;

	if (rkcg->rkcg_join_state != RD_KAFKA_CGRP_JOIN_STATE_WAIT_SYNC) {
		rd_kafka_dbg(rkb->rkb_rk, CGRP, "SYNCGROUP",
			     "SyncGroup response: discarding outdated request "
			     "(now in join-state %s)",
			     rd_kafka_cgrp_join_state_names[rkcg->
							    rkcg_join_state]);
		return;
	}

        if (err) {
                ErrorCode = err;
                goto err;
        }

        rd_kafka_buf_read_i16(rkbuf, &ErrorCode);
        rd_kafka_buf_read_bytes(rkbuf, &MemberState);

err:
        actions = rd_kafka_err_action(rkb, ErrorCode, request,
				      RD_KAFKA_ERR_ACTION_END);

        if (actions & RD_KAFKA_ERR_ACTION_REFRESH) {
                /* Re-query for coordinator */
                rd_kafka_cgrp_op(rkcg, NULL, RD_KAFKA_NO_REPLYQ,
				 RD_KAFKA_OP_COORD_QUERY,
                                 ErrorCode);
                /* FALLTHRU */
        }

        if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
                if (rd_kafka_buf_retry(rkb, request))
                        return;
                /* FALLTHRU */
        }

        rd_kafka_dbg(rkb->rkb_rk, CGRP, "SYNCGROUP",
                     "SyncGroup response: %s (%d bytes of MemberState data)",
                     rd_kafka_err2str(ErrorCode),
                     RD_KAFKAP_BYTES_LEN(&MemberState));

        if (ErrorCode == RD_KAFKA_RESP_ERR__DESTROY)
                return; /* Termination */

        rd_kafka_cgrp_handle_SyncGroup(rkcg, rkb, ErrorCode, &MemberState);

        return;

 err_parse:
        ErrorCode = rkbuf->rkbuf_err;
        goto err;
}
/**
 * Handler for SyncGroup responses
 * opaque must be the cgrp handle.
 */
Handler for SyncGroup responses
opaque must be the cgrp handle.
Handler for SyncGroup responses opaque must be the cgrp handle .
void rd_kafka_handle_SyncGroup (rd_kafka_t *rk,
				rd_kafka_broker_t *rkb,
                                rd_kafka_resp_err_t err,
                                rd_kafka_buf_t *rkbuf,
                                rd_kafka_buf_t *request,
                                void *opaque) {
        rd_kafka_cgrp_t *rkcg = opaque;
        const int log_decode_errors = LOG_ERR;
        int16_t ErrorCode = 0;
        rd_kafkap_bytes_t MemberState = RD_ZERO_INIT;
        int actions;
	if (rkcg->rkcg_join_state != RD_KAFKA_CGRP_JOIN_STATE_WAIT_SYNC) {
		rd_kafka_dbg(rkb->rkb_rk, CGRP, "SYNCGROUP",
			     "SyncGroup response: discarding outdated request "
			     "(now in join-state %s)",
			     rd_kafka_cgrp_join_state_names[rkcg->
							    rkcg_join_state]);
		return;
	}
        if (err) {
                ErrorCode = err;
                goto err;
        }
        rd_kafka_buf_read_i16(rkbuf, &ErrorCode);
        rd_kafka_buf_read_bytes(rkbuf, &MemberState);
err:
        actions = rd_kafka_err_action(rkb, ErrorCode, request,
				      RD_KAFKA_ERR_ACTION_END);
        if (actions & RD_KAFKA_ERR_ACTION_REFRESH) {
                rd_kafka_cgrp_op(rkcg, NULL, RD_KAFKA_NO_REPLYQ,
				 RD_KAFKA_OP_COORD_QUERY,
                                 ErrorCode);
        }
        if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
                if (rd_kafka_buf_retry(rkb, request))
                        return;
        }
        rd_kafka_dbg(rkb->rkb_rk, CGRP, "SYNCGROUP",
                     "SyncGroup response: %s (%d bytes of MemberState data)",
                     rd_kafka_err2str(ErrorCode),
                     RD_KAFKAP_BYTES_LEN(&MemberState));
        if (ErrorCode == RD_KAFKA_RESP_ERR__DESTROY)
                return; 
        rd_kafka_cgrp_handle_SyncGroup(rkcg, rkb, ErrorCode, &MemberState);
        return;
 err_parse:
        ErrorCode = rkbuf->rkbuf_err;
        goto err;
}
void rd_kafka_handle_SyncGroup ( rd_kafka_t * rk , rd_kafka_broker_t * rkb , rd_kafka_resp_err_t err , rd_kafka_buf_t * rkbuf , rd_kafka_buf_t * request , void * opaque ) { rd_kafka_cgrp_t * rkcg = opaque ; const int log_decode_errors = LOG_ERR ; int16_t ErrorCode = 0 ; rd_kafkap_bytes_t MemberState = RD_ZERO_INIT ; int actions ; if ( rkcg -> rkcg_join_state != RD_KAFKA_CGRP_JOIN_STATE_WAIT_SYNC ) { rd_kafka_dbg ( rkb -> rkb_rk , CGRP , " " , " " " " , rd_kafka_cgrp_join_state_names [ rkcg -> rkcg_join_state ] ) ; return ; } if ( err ) { ErrorCode = err ; goto err ; } rd_kafka_buf_read_i16 ( rkbuf , & ErrorCode ) ; rd_kafka_buf_read_bytes ( rkbuf , & MemberState ) ; err : actions = rd_kafka_err_action ( rkb , ErrorCode , request , RD_KAFKA_ERR_ACTION_END ) ; if ( actions & RD_KAFKA_ERR_ACTION_REFRESH ) { rd_kafka_cgrp_op ( rkcg , NULL , RD_KAFKA_NO_REPLYQ , RD_KAFKA_OP_COORD_QUERY , ErrorCode ) ; } if ( actions & RD_KAFKA_ERR_ACTION_RETRY ) { if ( rd_kafka_buf_retry ( rkb , request ) ) return ; } rd_kafka_dbg ( rkb -> rkb_rk , CGRP , " " , " " , rd_kafka_err2str ( ErrorCode ) , RD_KAFKAP_BYTES_LEN ( & MemberState ) ) ; if ( ErrorCode == RD_KAFKA_RESP_ERR__DESTROY ) return ; rd_kafka_cgrp_handle_SyncGroup ( rkcg , rkb , ErrorCode , & MemberState ) ; return ; err_parse : ErrorCode = rkbuf -> rkbuf_err ; goto err ; }
Handler for SyncGroup responses
 opaque must be the cgrp handle.
Handler for SyncGroup responses opaque must be the cgrp handle .
/* Re-query for coordinator */ /* FALLTHRU */ /* FALLTHRU */ /* Termination */