Kafka Consumer

Apache Kafka version 1.1.0

The Java version of Consumer Client

First look at the interface declaration:

/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE  file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this File to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the  License. */
package org.apache.kafka.clients.consumer;

import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.io.Closeable;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

/ * * *@see KafkaConsumer
 * @see MockConsumer
 */
public interface Consumer<K.V> extends Closeable {

    / * * *@see KafkaConsumer#assignment()
     */
    public Set<TopicPartition> assignment(a);

    / * * *@see KafkaConsumer#subscription()
     */
    public Set<String> subscription(a);

    / * * *@see KafkaConsumer#subscribe(Collection)
     */
    public void subscribe(Collection<String> topics);

    / * * *@see KafkaConsumer#subscribe(Collection, ConsumerRebalanceListener)
     */
    public void subscribe(Collection<String> topics, ConsumerRebalanceListener callback);

    / * * *@see KafkaConsumer#assign(Collection)
     */
    public void assign(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#subscribe(Pattern, ConsumerRebalanceListener)
    */
    public void subscribe(Pattern pattern, ConsumerRebalanceListener callback);

    / * * *@see KafkaConsumer#subscribe(Pattern)
    */
    public void subscribe(Pattern pattern);

    / * * *@see KafkaConsumer#unsubscribe()
     */
    public void unsubscribe(a);

    / * * *@see KafkaConsumer#poll(long)
     */
    public ConsumerRecords<K, V> poll(long timeout);

    / * * *@see KafkaConsumer#commitSync()
     */
    public void commitSync(a);

    / * * *@see KafkaConsumer#commitSync(Map)
     */
    public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);

    / * * *@see KafkaConsumer#commitAsync()
     */
    public void commitAsync(a);

    / * * *@see KafkaConsumer#commitAsync(OffsetCommitCallback)
     */
    public void commitAsync(OffsetCommitCallback callback);

    / * * *@see KafkaConsumer#commitAsync(Map, OffsetCommitCallback)
     */
    public void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback);

    / * * *@see KafkaConsumer#seek(TopicPartition, long)
     */
    public void seek(TopicPartition partition, long offset);

    / * * *@see KafkaConsumer#seekToBeginning(Collection)
     */
    public void seekToBeginning(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#seekToEnd(Collection)
     */
    public void seekToEnd(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#position(TopicPartition)
     */
    public long position(TopicPartition partition);

    / * * *@see KafkaConsumer#committed(TopicPartition)
     */
    public OffsetAndMetadata committed(TopicPartition partition);

    / * * *@see KafkaConsumer#metrics()
     */
    public Map<MetricName, ? extends Metric> metrics();

    / * * *@see KafkaConsumer#partitionsFor(String)
     */
    public List<PartitionInfo> partitionsFor(String topic);

    / * * *@see KafkaConsumer#listTopics()
     */
    public Map<String, List<PartitionInfo>> listTopics();

    / * * *@see KafkaConsumer#paused()
     */
    public Set<TopicPartition> paused(a);

    / * * *@see KafkaConsumer#pause(Collection)
     */
    public void pause(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#resume(Collection)
     */
    public void resume(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#offsetsForTimes(java.util.Map)
     */
    public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch);

    / * * *@see KafkaConsumer#beginningOffsets(java.util.Collection)
     */
    public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#endOffsets(java.util.Collection)
     */
    public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#close()
     */
    public void close(a);

    / * * *@see KafkaConsumer#close(long, TimeUnit)
     */
    public void close(long timeout, TimeUnit unit);

    / * * *@see KafkaConsumer#wakeup()
     */
    public void wakeup(a);

}

Copy the code

Consumer apis fall into the following categories:

1. Subscribe to a list of specified topics or topics that match a pattern and return the list of subscribed topics and unsubscribe

Subscribe to a list of topics or topics that match a pattern

    
        /**
         * Subscribe to the given list of topics to get dynamically assigned partitions.
         * <b>Topic subscriptions are not incremental. This list will replace the current
         * assignment (if there is one).</b> It is not possible to combine topic subscription with group management
         * with manual partition assignment through {@link #assign(Collection)}.
         *
         * If the given list of topics is empty, it is treated the same as {@link #unsubscribe()}.
         *
         * <p>
         * This is a short-hand for {@link #subscribe(Collection, ConsumerRebalanceListener)}, which
         * uses a no-op listener. If you need the ability to seek to particular offsets, you should prefer
         * {@link #subscribe(Collection, ConsumerRebalanceListener)}, since group rebalances will cause partition offsets
         * to be reset. You should also provide your own listener if you are doing your own offset
         * management since the listener gives you an opportunity to commit offsets before a rebalance finishes.
         *
         * @param topics The list of topics to subscribe to
         * @throws IllegalArgumentException If topics is null or contains null or empty elements
         * @throws IllegalStateException If {@code subscribe()} is called previously with pattern, or assign is called
         *                               previously (without a subsequent call to {@link #unsubscribe()}), or if not
         *                               configured at-least one partition assignment strategy
     */
    public void subscribe(Collection<String> topics);

    / * * *@see KafkaConsumer#subscribe(Collection, ConsumerRebalanceListener)
     */
    public void subscribe(Collection<String> topics, ConsumerRebalanceListener callback);
        / * * *@see KafkaConsumer#subscribe(Pattern, ConsumerRebalanceListener)
    */
    public void subscribe(Pattern pattern, ConsumerRebalanceListener callback);

    / * * *@see KafkaConsumer#subscribe(Pattern)
    */
    public void subscribe(Pattern pattern);
Copy the code

// Returns a list of subscribed topics

    / * * *@see KafkaConsumer#subscription()
     */
    public Set<String> subscription(a);
Copy the code

// Unsubscribe

    / * * *@see KafkaConsumer#unsubscribe()
     */
    public void unsubscribe(a);
Copy the code

2. Distribution consumes the assigned topic and gets the assigned Topic partition

// Allocate consumption to the specified topic

    / * * *@see KafkaConsumer#assign(Collection)
     */
    public void assign(Collection<TopicPartition> partitions);
Copy the code

// Get the assigned Topic partition

    / * * *@see KafkaConsumer#assignment()
     */
    public Set<TopicPartition> assignment(a);
Copy the code

3. Pull the data

    / * * *@see KafkaConsumer#poll(long)
     */
    public ConsumerRecords<K, V> poll(long timeout);
Copy the code

4. Submit offsets

    / * * *@see KafkaConsumer#commitSync()
     */
    public void commitSync(a);

    / * * *@see KafkaConsumer#commitSync(Map)
     */
    public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);

    / * * *@see KafkaConsumer#commitAsync()
     */
    public void commitAsync(a);

    / * * *@see KafkaConsumer#commitAsync(OffsetCommitCallback)
     */
    public void commitAsync(OffsetCommitCallback callback);

    / * * *@see KafkaConsumer#commitAsync(Map, OffsetCommitCallback)
     */
    public void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback);
Copy the code

5. Locate and obtain offsets

    / * * *@see KafkaConsumer#seek(TopicPartition, long)
     */
    public void seek(TopicPartition partition, long offset);

    / * * *@see KafkaConsumer#seekToBeginning(Collection)
     */
    public void seekToBeginning(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#seekToEnd(Collection)
     */
    public void seekToEnd(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#position(TopicPartition)
     */
    public long position(TopicPartition partition);
        / * * *@see KafkaConsumer#committed(TopicPartition)
     */
    public OffsetAndMetadata committed(TopicPartition partition);
        / * * *@see KafkaConsumer#offsetsForTimes(java.util.Map)
     */
    public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch);
        / * * *@see KafkaConsumer#beginningOffsets(java.util.Collection)
     */
    public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#endOffsets(java.util.Collection)
     */
    public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions);
Copy the code

6. Get topic and topic partition information

    / * * *@see KafkaConsumer#partitionsFor(String)
     */
    public List<PartitionInfo> partitionsFor(String topic);

    / * * *@see KafkaConsumer#listTopics()
     */
    public Map<String, List<PartitionInfo>> listTopics();
Copy the code

7. Control consumption behavior (pause, resume, wake up, close)

    / * * *@see KafkaConsumer#paused()
     */
    public Set<TopicPartition> paused(a);

    / * * *@see KafkaConsumer#pause(Collection)
     */
    public void pause(Collection<TopicPartition> partitions);

    / * * *@see KafkaConsumer#resume(Collection)
     */
    public void resume(Collection<TopicPartition> partitions);
        / * * *@see KafkaConsumer#close()
     */
    public void close(a);

    / * * *@see KafkaConsumer#close(long, TimeUnit)
     */
    public void close(long timeout, TimeUnit unit);

    / * * *@see KafkaConsumer#wakeup()
     */
    public void wakeup(a);
Copy the code

8. Get consumption indicators

    / * * *@see KafkaConsumer#metrics()
     */
    public Map<MetricName, ? extends Metric> metrics();
Copy the code

Initialize KafkaConsumer

KafkaConsumer has the following constructors:

The most common one is KafkaConsumer(Properties), where the passed Properties object is converted to the internal ConsumerConfig object, KafkaConsumer(ConsumerConfig config, Deserializer

keyDeserializer, Deserializer

valueDeserializer) So focus on the conversion of Properties to the ConsumerConfig object, see what configuration parameters are available, and then explore the logic inside the constructor.

First look at the ConsumerConfig class, where you can see the arguments required to construct KafkaConsumer:

The KafkaConsumer build steps are as follows:

1) Set clientId

First fetch the key from ConsumerConfig as the corresponding value of client.id. If this is not set, generate one, incrementing the sequence number with “consumer-” + CONSUMER_CLIENT_ID_SEQUENCE.

2) Set groupId

Select key from ConsumerConfig for group.id

3) Set requestTimeoutMs parameters

Take the value of request.timeout.ms from ConsumerConfig

4) Set the sessionTimeOutMs parameter

Retrieve key from ConsumerConfig for session.timeout.ms

Set the fetchMaxWaitMs parameter

Fetch the key from ConsumerConfig as fetch. Max.wait. ms

6) Check whether requestTimeoutMs is greater than sessionTimeOutMs and fetchMaxWaitMs

7) Set time

8) Build Metrics

9) Set retryBackoffMs

Fetches the value of retry.backoff.ms from ConsumerConfig

10) Build the ConsumerInterceptor list

11) Set the keyDeserializer

12) Set valueDeserializer

13) Configure ClusterResourceListeners

14) Build MetaData instances

This step simply creates a Metadata object and sets the parameters, but does not access the Kafka cluster to actually fetch the Metadata

15) Generate the InetSocketAddress according to the bootstrap.servers parameter

16) Update Metadata

The Cluster object consists of nodes, topics, topic partitions, and index relationships between them. No metadata is actually retrieved here, but Node information is supplemented by the bootstrap.servers parameter

17) Build ChannelBuilder

18) Obtain IsolationLevel

Take key from ConsumerConfig to isolation.level

19) Get the heartbeatIntervalMs parameter

20) Build NetworkClient

This is a key step in building KafkaConsumer, where you create a new Selector object (NIO Selector).

21) Construct the ConsumerNetworkClient object

22) Get OffsetResetStrategy

23) Construct PartitionAssignor

Create RangeAssignor instance

24) Build the ConsumerCoordinator

25) Build Fetcher

The key objects are Fetcher, ConsumerCoordinator, ConsumerNetworkClient, and NetworkClient.

Subscribe Topic list