kafka-consumer-internals-Fetcher.adoc

Fetcher

Fetcher is created exclusively when KafkaConsumer is created.

Figure 1. Fetcher and KafkaConsumer

KafkaConsumer uses the following configuration properties to create a Fetcher:

• fetch.min.bytes for minimum number of bytes

• fetch.max.bytes for maximum number of bytes

• fetch.max.wait.ms for maximum wait time

• max.partition.fetch.bytes for fetch size

• max.poll.records for how many records to poll

• check.crcs for flag to check CRC or not

• retry.backoff.ms for retry backoff (in milliseconds)

• request.timeout.ms for request timeout (in milliseconds)

Table 1. Fetcher’s Internal Properties (e.g. Registries and Counters) (in alphabetical order)

Name	Description
client	ConsumerNetworkClient that is given when Fetcher is created.

Creating Fetcher Instance

Fetcher takes the following when created:

• ConsumerNetworkClient

• Minimum number of bytes

• Maximum number of bytes

• Maximum wait time

• Fetch size

• How many records to poll

• Flag to check CRC or not

• Deserializer for keys

• Deserializer for values

• Metadata

• SubscriptionState

• Metrics

• FetcherMetricsRegistry

• Time

• Retry backoff (in milliseconds)

• Request timeout (in milliseconds)

• IsolationLevel

Fetcher initializes the internal registries and counters.

Fetcher registers itself with SubscriptionState as a listener to receive notifications about…​FIXME

sendFetches Method

int sendFetches()

sendFetches…​FIXME

Note	sendFetches is used when KafkaConsumer is requested to poll for records.

sendMetadataRequest Internal Method

RequestFuture<ClientResponse> sendMetadataRequest(MetadataRequest.Builder request)

Internally, sendMetadataRequest requests ConsumerNetworkClient for the least loaded node.

With the node, sendMetadataRequest requests ConsumerNetworkClient to send the request to the node.

When no node was found, sendMetadataRequest returns a RequestFuture with NoAvailableBrokersException.

Note	sendMetadataRequest is used exclusively when Fetcher is requested for a topic metadata.

beginningOffsets Method

Caution

FIXME

retrieveOffsetsByTimes Method

Caution

FIXME

Getting Topic Metadata — getTopicMetadata Method

Map<String, List<PartitionInfo>> getTopicMetadata(
  MetadataRequest.Builder request,
  long timeout)

Internally, getTopicMetadata sends the metadata request and requests ConsumerNetworkClient to poll until it finishes successfully or timeout expires.

After poll finishes, getTopicMetadata takes the cluster information from MetadataResponse.

When MetadataResponse is successful, getTopicMetadata takes topics (from Cluster) and requests Cluster for available partitions for every topic.

In the end, getTopicMetadata creates a collection of topic and partitions pairs.

Caution

FIXME Describe the failure path

Note	getTopicMetadata is used when: Fetcher is requested to find metadata for all topics KafkaConsumer is requested to find partitions for a topic

Finding Metadata for All Topics — getAllTopicMetadata Method

Map<String, List<PartitionInfo>> getAllTopicMetadata(long timeout)

getAllTopicMetadata gets topic metadata specifying no topics (which means all topics available).

Note	getAllTopicMetadata is used exclusively when KafkaConsumer requests metadata for all topics.

sendListOffsetRequest Internal Method

RequestFuture<ListOffsetResult> sendListOffsetRequest(
  final Node node,
  final Map<TopicPartition, ListOffsetRequest.PartitionData> timestampsToSearch,
  boolean requireTimestamp)

sendListOffsetRequest…​FIXME

Note	sendListOffsetRequest is used when…​FIXME

fetchOffsetsByTimes Internal Method

ListOffsetResult fetchOffsetsByTimes(
  Map<TopicPartition, Long> timestampsToSearch,
  Timer timer,
  boolean requireTimestamps)

fetchOffsetsByTimes…​FIXME

Note	fetchOffsetsByTimes is used when…​FIXME