ConsumerNetworkClient

ConsumerNetworkClient is a high-level Kafka consumer that…FIXME

ConsumerNetworkClient is created for KafkaConsumer and AdminClient.

Tip	Enable `ALL` logging level for `org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient` logger to see what happens inside. Add the following line to `log4j.properties`: `log4j.logger.org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient=ALL` Refer to Logging.

`checkDisconnects` Internal Method

void checkDisconnects(long now)

checkDisconnects…FIXME

Note	`checkDisconnects` is used exclusively when `ConsumerNetworkClient` is requested to poll.

"Sending" Request to Broker — `send` Method

RequestFuture<ClientResponse> send(Node node, AbstractRequest.Builder<?> requestBuilder)

send creates a RequestFutureCompletionHandler and requests the KafkaClient for a new ClientRequest (with the RequestFutureCompletionHandler and expecting a response).

send records the new ClientRequest with the node in unsent internal registry.

send requests the KafkaClient to wake up.

Note	`send` is used…FIXME

`wakeup` Method

void wakeup()

wakeup prints out the following DEBUG message to the logs:

Received user wakeup

wakeup turns the internal wakeup flag on and requests KafkaClient to wakeup

Note	`wakeup` is used when…FIXME

Updating Cluster Metadata (and Possibly Blocking Until It Refreshes) — `ensureFreshMetadata` Method

void ensureFreshMetadata()

ensureFreshMetadata waits for metadata update when Metadata was requested for update or time to the next update is now.

Note	`ensureFreshMetadata` is used when `ConsumerCoordinator` does onJoinComplete, performAssignment and poll.

`pendingRequestCount` Method

int pendingRequestCount()
int pendingRequestCount(Node node)

pendingRequestCount…FIXME

Note	`pendingRequestCount` is used when…FIXME

Least-Loaded Kafka Node — `leastLoadedNode` Method

Node leastLoadedNode()

leastLoadedNode…FIXME

Note	`leastLoadedNode` is used when: `AbstractCoordinator` is requested to lookupCoordinator `Fetcher` is requested to sendMetadataRequest

`poll` Method

void poll(
  RequestFuture<?> future) (1)
boolean poll(
  RequestFuture<?> future,
  Timer timer)
void poll(
  Timer timer) (2)
void poll(
  Timer timer,
  PollCondition pollCondition) (3)
void poll(
  Timer timer,
  PollCondition pollCondition,
  boolean disableWakeup)

Used in tests (and uses a maximum timeout)
Uses uninitialized PollCondition (null)
Turns disableWakeup off (false)

poll…FIXME

Note

poll is used when:

AbstractCoordinator is requested to ensureCoordinatorReady and joinGroupIfNeeded
ConsumerNetworkClient is requested to pollNoWakeup, awaitMetadataUpdate, and awaitPendingRequests
Fetcher is requested to getTopicMetadata and fetchOffsetsByTimes
KafkaConsumer is requested to pollForFetches and position
ConsumerCoordinator is requested to close, fetchCommittedOffsets and commitOffsetsSync
BrokerApiVersionsCommand tool is executed (and creates a AdminClient)
Kafka Connect’s WorkerCoordinator is requested to poll

Waiting for Metadata Update — `awaitMetadataUpdate` Method

boolean awaitMetadataUpdate(long timeout)

awaitMetadataUpdate…FIXME

Note	`awaitMetadataUpdate` is used when…FIXME

`awaitPendingRequests` Method

boolean awaitPendingRequests(Node node, Timer timer)

awaitPendingRequests…FIXME

Note	`awaitPendingRequests` is used when…FIXME

`pollNoWakeup` Method

void pollNoWakeup()

pollNoWakeup…FIXME

Note	`pollNoWakeup` is used when: `KafkaConsumer` is requested to poll for records `AbstractCoordinator` is requested to maybeLeaveGroup `HeartbeatThread` is requested to run `ConsumerCoordinator` is requested to commitOffsetsAsync

Creating ConsumerNetworkClient Instance

ConsumerNetworkClient takes the following when created:

LogContext
KafkaClient
Metadata
Time
retryBackoffMs
requestTimeoutMs

ConsumerNetworkClient initializes the internal registries and counters.

`trySend` Internal Method

long trySend(long now)

trySend…FIXME

Note	`trySend` is used exclusively when `ConsumerNetworkClient` is requested to poll.

Initiating Connector to Kafka Node — `tryConnect` Method

void tryConnect(Node node)

tryConnect simply requests the KafkaClient to initiate a connection to the given broker Node.

Note	`tryConnect` is used exclusively when `FindCoordinatorResponseHandler` is requested to onSuccess.

`handlePendingDisconnects` Internal Method

void handlePendingDisconnects()

handlePendingDisconnects…FIXME

Note	`handlePendingDisconnects` is used exclusively when `ConsumerNetworkClient` is requested to poll.

`maybeTriggerWakeup` Method

void maybeTriggerWakeup()

maybeTriggerWakeup…FIXME

Note	`maybeTriggerWakeup` is used when…FIXME

`transmitSends` Method

void transmitSends()

transmitSends…FIXME

Note	`transmitSends` is used when `KafkaConsumer` is requested to poll.

Internal Properties

Name Description

Name	Description
`pendingCompletion`
`unsent`	`UnsentRequests` with pending requests per node that have not been sent yet (i.e. awaiting transmission).

pendingCompletion

unsent

UnsentRequests with pending requests per node that have not been sent yet (i.e. awaiting transmission).

ConsumerNetworkClient