const rd_kafkap_str_t *rkcg_group_id;

rd_kafkap_str_t *rkcg_member_id; /* Last assigned MemberId */

rd_kafkap_str_t *rkcg_group_instance_id;

const rd_kafkap_str_t *rkcg_client_id;

rd_kafkap_str_t *rkcg_client_rack;

enum {

/* Init state */

RD_KAFKA_CGRP_STATE_INIT,

/* Cgrp has been stopped. This is a final state */

RD_KAFKA_CGRP_STATE_TERM,

/* Query for group coordinator */

RD_KAFKA_CGRP_STATE_QUERY_COORD,

/* Outstanding query, awaiting response */

RD_KAFKA_CGRP_STATE_WAIT_COORD,

/* Wait ack from assigned cgrp manager broker thread */

RD_KAFKA_CGRP_STATE_WAIT_BROKER,

/* Wait for manager broker thread to connect to broker */

RD_KAFKA_CGRP_STATE_WAIT_BROKER_TRANSPORT,

/* Coordinator is up and manager is assigned. */

RD_KAFKA_CGRP_STATE_UP,

} rkcg_state;

rd_ts_t rkcg_ts_statechange; /* Timestamp of last

* state change. */

enum {

/* all: join or rejoin, possibly with an existing assignment. */

RD_KAFKA_CGRP_JOIN_STATE_INIT,

/* all: JoinGroupRequest sent, awaiting response. */

RD_KAFKA_CGRP_JOIN_STATE_WAIT_JOIN,

/* all: MetadataRequest sent, awaiting response.

* While metadata requests may be issued at any time,

* this state is only set upon a proper (re)join. */

RD_KAFKA_CGRP_JOIN_STATE_WAIT_METADATA,

/* Follower: SyncGroupRequest sent, awaiting response. */

RD_KAFKA_CGRP_JOIN_STATE_WAIT_SYNC,

/* all: waiting for application to call *_assign() */

RD_KAFKA_CGRP_JOIN_STATE_WAIT_ASSIGN_CALL,

/* all: waiting for application to call *_unassign() */

RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN_CALL,

/* all: waiting for full assignment to decommission */

RD_KAFKA_CGRP_JOIN_STATE_WAIT_UNASSIGN_TO_COMPLETE,

/* all: waiting for partial assignment to decommission */

RD_KAFKA_CGRP_JOIN_STATE_WAIT_INCR_UNASSIGN_TO_COMPLETE,

/* all: synchronized and assigned

* may be an empty assignment. */

RD_KAFKA_CGRP_JOIN_STATE_STEADY,

} rkcg_join_state;

/* State when group leader */

struct {

rd_kafka_group_member_t *members;

int member_cnt;

} rkcg_group_leader;

rd_kafka_q_t *rkcg_q; /* Application poll queue */

rd_kafka_q_t *rkcg_ops; /* Manager ops queue */

rd_kafka_q_t *rkcg_wait_coord_q; /* Ops awaiting coord */

int rkcg_flags;

#define RD_KAFKA_CGRP_F_TERMINATE 0x1 /* Terminate cgrp (async) */

#define RD_KAFKA_CGRP_F_LEAVE_ON_UNASSIGN_DONE \

0x8 /* Send LeaveGroup when \

* unassign is done */

#define RD_KAFKA_CGRP_F_SUBSCRIPTION \

0x10 /* If set: \

* subscription \

* else: \

* static assignment */

#define RD_KAFKA_CGRP_F_HEARTBEAT_IN_TRANSIT \

0x20 /* A Heartbeat request \

* is in transit, dont \

* send a new one. */

#define RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION \

0x40 /* Subscription contains \

* wildcards. */

#define RD_KAFKA_CGRP_F_WAIT_LEAVE \

0x80 /* Wait for LeaveGroup \

* to be sent. \

* This is used to stall \

* termination until \

* the LeaveGroupRequest \

* is responded to, \

* otherwise it risks \

* being dropped in the \

* output queue when \

* the broker is destroyed. \

*/

#define RD_KAFKA_CGRP_F_MAX_POLL_EXCEEDED \

0x100 /**< max.poll.interval.ms \

* was exceeded and we \

* left the group. \

* Do not rejoin until \

* the application has \

* polled again. */

rd_interval_t rkcg_coord_query_intvl; /* Coordinator query intvl*/

rd_interval_t rkcg_heartbeat_intvl; /* Heartbeat intvl */

rd_kafka_timer_t rkcg_serve_timer; /* Timer for next serve. */

int rkcg_heartbeat_intvl_ms; /* KIP 848: received

* heartbeat interval in

* milliseconds */

rd_interval_t rkcg_join_intvl; /* JoinGroup interval */

rd_interval_t rkcg_timeout_scan_intvl; /* Timeout scanner */

rd_ts_t rkcg_ts_session_timeout; /**< Absolute session

* timeout enforced by

* the consumer, this

* value is updated on

* Heartbeat success,

* etc. */

rd_kafka_resp_err_t rkcg_last_heartbeat_err; /**< Last Heartbeat error,

* used for logging. */

TAILQ_HEAD(, rd_kafka_topic_s) rkcg_topics; /* Topics subscribed to */

rd_list_t rkcg_toppars; /* Toppars subscribed to*/

int32_t rkcg_generation_id; /* Current generation id (classic)

* or member epoch (consumer). */

rd_kafka_assignor_t *rkcg_assignor; /**< The current partition

* assignor. used by both

* leader and members. */

void *rkcg_assignor_state; /**< current partition

* assignor state */

int32_t rkcg_coord_id; /**< Current coordinator id,

* or -1 if not known. */

rd_kafka_group_protocol_t

rkcg_group_protocol; /**< Group protocol to use */

rd_kafkap_str_t *rkcg_group_remote_assignor; /**< Group remote

* assignor to use */

rd_kafka_broker_t *rkcg_curr_coord; /**< Current coordinator

* broker handle, or NULL.

* rkcg_coord's nodename is

* updated to this broker's

* nodename when there is a

* coordinator change. */

rd_kafka_broker_t *rkcg_coord; /**< The dedicated coordinator

* broker handle.

* Will be updated when the

* coordinator changes. */

int16_t rkcg_wait_resp; /**< Awaiting response for this

* ApiKey.

* Makes sure only one

* JoinGroup or SyncGroup

* request is outstanding.

* Unset value is -1. */

/** Current subscription */

rd_kafka_topic_partition_list_t *rkcg_subscription;

/** The actual topics subscribed (after metadata+wildcard matching).

* Sorted. */

rd_list_t *rkcg_subscribed_topics; /**< (rd_kafka_topic_info_t *) */

/** Subscribed topics that are errored/not available. */

rd_kafka_topic_partition_list_t *rkcg_errored_topics;

/** If a SUBSCRIBE op is received during a COOPERATIVE rebalance,

* actioning this will be postponed until after the rebalance

* completes. The waiting subscription is stored here. */

rd_kafka_topic_partition_list_t *rkcg_next_subscription;

/**

* Subscription regex pattern. All the provided regex patterns are

* stored as a single string with each pattern separated by '|'.

*

* Only applicable for the consumer protocol introduced in KIP-848.

*

* rkcg_subscription = rkcg_subscription_topics +

* rkcg_subscription_regex

*/

rd_kafkap_str_t *rkcg_subscription_regex;

/**

* Full topic names extracted out from the rkcg_subscription.

*

* Only applicable for the consumer protocol introduced in KIP-848.

*

* For the consumer protocol, this field doesn't include regex

* subscriptions. For that please refer `rkcg_subscription_regex`

*

* rkcg_subscription = rkcg_subscription_topics +

* rkcg_subscription_regex

*/

rd_kafka_topic_partition_list_t *rkcg_subscription_topics;

/** If a (un)SUBSCRIBE op is received during a COOPERATIVE rebalance,

* actioning this will be posponed until after the rebalance

* completes. This flag is used to signal a waiting unsubscribe

* operation. Mutually exclusive with rkcg_next_subscription. */

rd_bool_t rkcg_next_unsubscribe;

/** Assignment considered lost */

rd_atomic32_t rkcg_assignment_lost;

/** Current assignment of partitions from last SyncGroup response.

* NULL means no assignment, else empty or non-empty assignment.

*

* This group assignment is the actual set of partitions that were

* assigned to our consumer by the consumer group leader and should

* not be confused with the rk_consumer.assignment which is the

* partitions assigned by the application using assign(), et.al.

*

* The group assignment and the consumer assignment are typically

* identical, but not necessarily since an application is free to

* assign() any partition, not just the partitions it is handed

* through the rebalance callback.

*

* Yes, this nomenclature is ambigious but has historical reasons,

* so for now just try to remember that:

* - group assignment == consumer group assignment.

* - assignment == actual used assignment, i.e., fetched partitions.

*

* @remark This list is always sorted.

*/

rd_kafka_topic_partition_list_t *rkcg_group_assignment;

/** The partitions to incrementally assign following a

* currently in-progress incremental unassign. */

rd_kafka_topic_partition_list_t *rkcg_rebalance_incr_assignment;

/** Current acked assignment, start with an empty list. */

rd_kafka_topic_partition_list_t *rkcg_current_assignment;

/** Assignment the is currently reconciling.

* Can be NULL in case there's no reconciliation ongoing. */

rd_kafka_topic_partition_list_t *rkcg_target_assignment;

/** Next assignment that will be reconciled once current

* reconciliation finishes. Can be NULL. */

rd_kafka_topic_partition_list_t *rkcg_next_target_assignment;

/** Number of backoff retries when expediting next heartbeat. */

int rkcg_expedite_heartbeat_retries;

/** Flags for KIP-848 state machine. */

int rkcg_consumer_flags;

/** Coordinator is waiting for an acknowledgement of currently reconciled

* target assignment. Cleared when an HB succeeds

* after reconciliation finishes. */

#define RD_KAFKA_CGRP_CONSUMER_F_WAIT_ACK 0x1

/** Member is sending an acknowledgement for a reconciled assignment */

#define RD_KAFKA_CGRP_CONSUMER_F_SENDING_ACK 0x2

/** A new subscription needs to be sent to the Coordinator. */

#define RD_KAFKA_CGRP_CONSUMER_F_SEND_NEW_SUBSCRIPTION 0x4

/** A new subscription is being sent to the Coordinator. */

#define RD_KAFKA_CGRP_CONSUMER_F_SENDING_NEW_SUBSCRIPTION 0x8

/** Consumer has subscribed at least once,

* if it didn't happen rebalance protocol is still

* considered NONE, otherwise it depends on the

* configured partition assignors. */

#define RD_KAFKA_CGRP_CONSUMER_F_SUBSCRIBED_ONCE 0x10

/** Send a complete request in next heartbeat */

#define RD_KAFKA_CGRP_CONSUMER_F_SEND_FULL_REQUEST 0x20

/** Member is fenced, need to rejoin */

#define RD_KAFKA_CGRP_CONSUMER_F_WAIT_REJOIN 0x40

/** Member is fenced, rejoining */

#define RD_KAFKA_CGRP_CONSUMER_F_WAIT_REJOIN_TO_COMPLETE 0x80

/** Serve pending assignments after heartbeat */

#define RD_KAFKA_CGRP_CONSUMER_F_SERVE_PENDING 0x100

/** Rejoin the group following a currently in-progress

* incremental unassign. */

rd_bool_t rkcg_rebalance_rejoin;

rd_ts_t rkcg_ts_last_err; /* Timestamp of last error

* propagated to application */

rd_kafka_resp_err_t rkcg_last_err; /* Last error propagated to

* application.

* This is for silencing

* same errors. */

rd_kafka_timer_t rkcg_offset_commit_tmr; /* Offset commit timer */

rd_kafka_timer_t rkcg_max_poll_interval_tmr; /**< Enforce the max

* poll interval. */

rd_kafka_t *rkcg_rk;

rd_kafka_op_t *rkcg_reply_rko; /* Send reply for op

* (OP_TERMINATE)

* to this rko's queue. */

rd_ts_t rkcg_ts_terminate; /* Timestamp of when

* cgrp termination was

* initiated. */

rd_atomic32_t rkcg_terminated; /**< Consumer has been closed */

rd_atomic32_t rkcg_subscription_version; /**< Subscription version */

/* Protected by rd_kafka_*lock() */

struct {

rd_ts_t ts_rebalance; /* Timestamp of

* last rebalance */

int rebalance_cnt; /* Number of

rebalances */

char rebalance_reason[256]; /**< Last rebalance

* reason */

int assignment_size; /* Partition count

* of last rebalance

* assignment */

} rkcg_c;

/* Timestamp of last rebalance start */

rd_ts_t rkcg_ts_rebalance_start;

rd_kafka_toppar_t *rktp,

rd_kafka_replyq_t replyq,

rd_kafka_op_type_t type,

rd_kafka_cgrp_t *rkcg,

const rd_kafka_topic_partition_list_t *offsets,

rd_bool_t set_offsets,

char *group_id;

int32_t generation_id;

char *member_id;

char *group_instance_id; /**< Optional (NULL) */