...

• Define the responsibilities of the polling thread and the background thread
• Define the communication interface and protocol between the two threads
• Define a Network layer over the NetworkClient, including how different requests are sent and handled
  • Define RequestManagers
  • Define RequestStates to handle retry backoff
• Redesign SubscriptionState
• Deprecate HeartbeatThread and move it to the background thread
• Define rebalance flow, including how JoinGroup and SyncGroup request is sent and how callbacks are triggered

• Refactor the KafkaConsumer API innards

• Address issues in these Jira tickets

Terms

Polling Thread

The client thread, or the polling thread, is where the thread user invokes the poll() operation.

Background Thread

Specifically, we use this generic term in the new design context to indicate the active process running in the background. It handles network communication, rebalancing, and heartbeat.

Heartbeat Thread

The background running thread in the current implementation (see KIP-62).

Channel

The communication interface between two threads. One channel delivers messages from the polling thread to the background thread. The other channel delivers messages submitted by the background thread to the polling thread.

Event

The primary communication medium for the polling thread is to interact with the background thread.

Design

The new consumer is asynchronous and event drive. Former means we are separating the responsibility between the polling thread and the background thread. The latter means the two threads communicate via events. The core components are:

• Application thread: user thread on which all Consumer API requests and callback handlers are executed.

• Network thread: client-internal thread on which network I/O related to heartbeat, coordinator, and rebalance is executed.

• Event handler: A communication interface between the application thread and network thread

• Application event queue: sends events to the network I/O thread from the application thread

• Background event queue: sends events from the network thread to the application thread

We will discuss the strategy for handling event results in the following section. In short, for async APIs like commitAsync(), CompletableFuture will be used to notify the completion of the events.

Application and Network Thread

The following diagram depicts the interaction between the application thread and the network thread:

Design

The new consumer is asynchronous and event drive. Former means we are separating the responsibility between the polling thread and the background thread. The latter means the two threads communicate via events. The core components are:

• Application thread: user thread on which all Consumer API requests and callback handlers are executed.

• Network thread: client-internal thread on which network I/O related to heartbeat, coordinator, and rebalance is executed.

• Event handler: A communication interface between the application thread and network thread

• Application event queue: sends events to the network I/O thread from the application thread

• Background event queue: sends events from the network thread to the application thread

• Application event processor: 

We will discuss the strategy for handling event results in the following section. In short, for async APIs like commitAsync(), CompletableFuture will be used to notify the completion of the events.

Application and Network Thread

The following diagram depicts the interaction between the application thread and the network thread:

The Consumer client object is here depicted in purple. In this design, instead of directly operating on the parameters given to the various APIs (subscribe(), poll(), commit(), etc.), the Consumer implementation packages the parameters as events that are enqueued on the application event queue.

The event handler, shown in green, executes on the network thread. In each loop of the network thread, events are read from the queue and processed.

If the Consumer API is a blocking call, the event passed from the application thread to the network thread will include an embedded CompleteableFuture. After enqueuing the event, the application thread will invoke Future.get(), effectively blocking itself until a result is provided. When the result for the Consumer API call is ready, the network thread will then invoke CompeteableFuture.complete() with the result, allowing the application thread to continue execution.

Submitting Client Requests

The following diagram displays the basic flow between the request managers, unsent request queue, and the NetworkClient: 

The Consumer client object is here depicted in purple. In this design, instead of directly operating on the parameters given to the various APIs (subscribe(), poll(), commit(), etc.), the Consumer implementation packages the parameters as events that are enqueued on the application event queue.

The event handler, shown in green, executes on the network thread. In each loop of the network thread, events are read from the queue and processed.

...

A request manager represents the logic and stated needed to issue a Kafka RPC requests and handle its response. A request manager may contain logic to handle more than one type of RPC.

In the network thread, we loop over each request manager, effectively asking it for any requests that it needs to send to the cluster. Note that during this step the network request is not sent. Instead, an unsent request object is created which contains the underlying request information. These "unsent requests" are added to a queue of pending unsent client requests. After all of these unsent requests are queued up, then they are forwarded for network I/O via the NetworkClient.send() method.

There are two benefits for this multi-step process:

1. It keeps the network I/O request and response management and lifecycle in one place, making the code easier to reason about
2. The request managers can implement deduplication and/or coalescing of requests

Terminologies:

• CB: Callbacks registered and invoked on the polling thread: commit callback, rebalance callback.
• rm: RequestManagers. e.g. Heartbeat, FindCoordinatorRequest.
• subscriptionState design is still under discussion

...

Application thread and its lifecycle

The polling thread handles API invocation and any responses from the background thread. Let's demonstrate its life cycle using the simple consumer use case (assign(), poll(), commitSync()) :

...

// Channel used to send events to the polling thread for client side execution/notification

private BlockingQueue<BackgroundEvent> queue;

abstract public class BackgroundEvent {
   private final BackgroundEventType eventType;
}

enum BackgroundEventType {
   ERROR,
   REVOKE_PARTITIONS,
   ASSIGN_PARTITIONS,
   FETCH_RESPONSE,
}

Rebalance [WIP]

One of the main reasons we are refactoring the KafkaConsumer is to satisfy the requirements of the new rebalance protocol introduced in KIP-848.

KIP-848 contains two assignment modes, server-side mode and client-side mode.  Both use the new Heartbeat API, the ConsumerGroupHeartbeat.

The server-side mode is simpler: the assignments are computed by the Group Coordinator, and the clients are only responsible for revoking and assigning the partitions.

If the user chooses to use the client-side assignor, the assignment will be computed by one of the member, and the assignment and revocation is done via the heartbeat as server side mode.

In the new design we will build the following components:

1. GroupState: keep track of the current state of the group, such as Generation, and the rebalance state.
2. HeartbeatRequestManager: A type of request manager that is responsible for calling the ConsumerGroupHeartbeat API
3. Assignment Manager: Manages partition assignments.

Rebalance Flow

New Consumer Group

1. The user invokes subscribe(). SubscriptionState is altered. A subscription state alters event is sent to the background thread.
2. The background thread processes the event and updates the GroupState to PREPARE.
3. HeartbeatRequestManager is polled. It checks the GroupState and determines it is time to send the heartbeat.
4. ConsumerGroupHeartbeatResponse received. Updated the GroupState to ASSIGN.
5. PartitionAssignmentManager is polled, and realize the GroupState is in ASSIGN. Trigger assignment computation:
6. [We might need another state here]
7. Once the assignment is computed, send an event to the client thread to invoke the rebalance callback.
8. Callback triggered; notify the background thread.
9. PartitionAssignmentManager is polled Transition to Complete.
10. [something needs to happen here]
11. Transition the GroupState to Stable.

GroupState

[UNJOINED, PREPARE, ASSIGN, COMPLETE, STABLE]

• UNJOINED: There's no rebalance. For the simple consumed use case, the GroupState remains in UNJOINED
• PREPARE: Sending the heartbeat and await the response
• ASSIGN: Assignment updated, client thread side callbacks are triggered, and await completion
• COMPLETE: Client thread callback completed and has notified the background thread.
• STABLE: stable group

Consumer group member state machine

It becomes clear when reading KIP-848 that the work of keeping the consumer group in proper state is fairly involved. We therefore turn our focus now to the logic needed for the consumer group member state machine (hereafter, CGMSM). 

Based on the user calling either assign() or subscribe(), a Consumer determines how topic partitions are to be assigned. If the user calls the subscribe() API, the Consumer knows that it is being directed to use Kafka's consumer group-based partition assignment. The use of assign() signifies the user's intention to manage the partition assignments from within the application via manual partition assignment. It is only in the former case that a CGMSM needs to be created.

Note that the necessary logic to establish a connection to the Kafka broker node acting as the group coordinator is outside the scope of the CGMSM logic.

In order to keep the size of a ConsumerGroupHeartbeatRequest smaller, KIP-848's description of the request schema states that some values are conditionally sent with the request only when they change on the client. These values include:

• InstanceId
• RackId
• RebalanceTimeoutMs
• SubscribedTopicNames
• SubscribedTopicRegex
• ServerAssignor
• ClientAssignors
• TopicPartitions

The following diagram provides a visual overview of the states and transitions for members of the consumer group:

...

The following description provides more clarity on the states that make up the CGMSM:

NEW

NEW is the initial state for a CGMSM upon its creation. The Consumer will remain in this state until the next pass of the background thread loop.

JOINING

A state of JOINING signifies that a Consumer wants to join a consumer group. On the next pass of the background thread, the Consumer will enter this state to begin communicating with the Kafka broker node that was elected as the group coordinator. A ConsumerGroupHeartbeatRequest will be sent to the coordinator with specific values in the request:

• MemberId is set to null
• MemberEpoch is set to the hard-coded value of 0

Since this is the first request to the coordinator, the CGMSM will include a ConsumerGroupHeartbeatRequest with all conditional values present. This includes setting TopicPartitions to null since there are no assigned partitions in this state.

Once the initial ConsumerGroupHeartbeatResponse is received successfully, the CGMSM will update its local MemberId and MemberEpoch based on the returned data. It will then transition to the JOINED state.

JOINED

The JOINED state simply indicates that the Consumer instance is known to the coordinator as a member of the group. It does not necessarily imply that it has been assigned any partitions. While in the JOINED state the CGMSM will periodically send requests to the coordinator at the needed cadence in order to maintain membership.

The CGMSM should transition back to the JOINING state if the ConsumerGroupHeartbeatResponse has an error of UNKNOWN_MEMBER_ID or FENCED_MEMBER_EPOCH. If either of those errors occur, the CGMSM will clear its "assigned" partition set (without any revocation), and transition to the JOINING set so that it rejoins the group with the same MemberId and the MemberEpoch of 0.

The CGMSM will transition into the ASSIGNING state when the ConsumerGroupHeartbeatResponse contains a non-null value for Assignment.

ASSIGNING

The ASSIGNING state is entered with the intention that the CGMSM will need to perform the assignment reconciliation process. As is done in the JOINED state, the CGMSM will continue to communicate with the coordinator via the heartbeat mechanism to maintain its membership.

The first action that is performed in this state is to update the CGMSM's value for the member epoch as provided in the ConsumerGroupHeartbeatResponse.

Next, the CGMSM performs a comparison between its current the assignment and the value of Assignment contained in the ConsumerGroupHeartbeatResponse. If the two assignments are equal, the CGMSM has reconciled the assignment successful and will transition back to the JOINED state. If they are not equal, the reconciliation process begins.

KIP-848 states that during reconciliation, partitions are revoked first and then assigned second, as two distinct steps.

Partition revocation involves:

1. Removing the partitions from the CGMSM's "assigned" set
2. Commits the offsets for the revoked partitions
3. Invokes ConsumerRebalanceListener.onPartitionsRevoked()

Partition assignment includes:

1. Adding the partitions to the CGMSM's "assigned" set
2. Invokes ConsumerPartitionAssignor.onAssignment(), if one is set
3. Invokes ConsumerRebalanceListener.onPartitionsAssigned()

Questions

1. Do we need to heartbeat between revocation and assignment?
2. Do we want to split up ASSIGNING into separate states REVOKING and ASSIGNING?

TERMINATING

TBD

TERMINATED

...

   ASSIGN_PARTITIONS,
   FETCH_RESPONSE,
}

Consumer API Internal Changes 

...

We will break the current fetcher into three parts to accommodate the asynchronous design, i.e., we need the background thread to send fetches autonomously and the polling thread to collect fetches when these fetches become available. We will have 3 separate classes here:

1. FetchSender: Responsible for sending fetches in the background thread
2. FetchHandler: Sitting in the polling thread's poll loop, processing the fetch response from the fetch event. 
3. FetchBuffer: This is the CompletedFetches in the old implementation. The use case prevents the FetchSender from sending too many fetches and causing memory issues. This will be removed once we implement the memory-based buffer.(KIP-81)

Consumer Poll Changes

We will remove the metadata logic from the consumer.poll(), so that the execution of the poll loop is much simplified. It mainly handles:

1. fetches
2. callback invocation
3. errors

ConsumerCoordinator and AbstractCoordinator

• New states will be introduced (see Rebalance States section above).  The main purpose is to make the background thread drive the poll, and letting the polling thread to invoke the callbacks.

• Remove HeartbeatThread. Therefore, we won't be starting the heartbeat thread.

  • It will still require a fetch event to poll heartbeat.  As only polling thread issues fetch events, and we want to respect the existing implementation.
• Timeouts and timers will be reevaluated and possibly removed.
• while loops will be reevaluated and possibly thrown out.  In the new implementation the coordinator will be non-blocking, and its states are managed by the background thread loop.

Timeout Policy

Consumer.poll() - user provide timeout

Coordinator rediscovery backoff: retry.backoff.ms

Coordinator discovery timeout: Currently uses the user-provided timeout in the consumer.poll(). Maybe we should use request.timeout.ms. And re-attempt in the next loop if failed

CommitOffsetSync: user provided

Rebalance State Timeout: maybe using the request timeout

, we need the background thread to send fetches autonomously and the polling thread to collect fetches when these fetches become available. We will have 3 separate classes here:

1. FetchSender: Responsible for sending fetches in the background thread
2. FetchHandler: Sitting in the polling thread's poll loop, processing the fetch response from the fetch event. 
3. FetchBuffer: This is the CompletedFetches in the old implementation. The use case prevents the FetchSender from sending too many fetches and causing memory issues. This will be removed once we implement the memory-based buffer.(KIP-81)

Consumer Poll Changes

We will remove the metadata logic from the consumer.poll(), so that the execution of the poll loop is much simplified. It mainly handles:

1. fetches
2. callback invocation
3. errors

ConsumerCoordinator and AbstractCoordinator

• New states will be introduced (see Rebalance States section above).  The main purpose is to make the background thread drive the poll, and letting the polling thread to invoke the callbacks.

• Remove HeartbeatThread. Therefore, we won't be starting the heartbeat thread.

  • It will still require a fetch event to poll heartbeat.  As only polling thread issues fetch events, and we want to respect the existing implementation.
• Timeouts and timers will be reevaluated and possibly removed.
• while loops will be reevaluated and possibly thrown out.  In the new implementation the coordinator will be non-blocking, and its states are managed by the background thread loop.

Timeout Policy

Please see Java client Consumer timeouts for more detail on timeouts.Is there a better way to configure session interval and heartbeat interval?

Compatibility

The new consumer should be backward compatible.

...

The refactor should have (almost) no regression or breaking changes upon merge. So user should be able to continue using the new client. draw.io DiagrambordertruediagramNameApplication and network threadssimpleViewerfalsewidthlinksautotbstyletoplboxtruediagramWidth923revision1