Kafka Consumer for Confluent Platform

An Apache Kafka® Consumer is a client application that subscribes to (reads and processes) events. This section provides an overview of the Kafka consumer and an introduction to the configuration settings for tuning.

Ready to get started?

The Kafka consumer works by issuing “fetch” requests to the brokers leading the partitions it wants to consume. The consumer offset is specified in the log with each request. The consumer receives back a chunk of log that contains all of the messages in that topic beginning from the offset position. The consumer has significant control over this position and can rewind it to re-consume data if desired.

Consumer groups

A consumer group is a set of consumers that cooperate to consume data from some topics. You set the group for a consumer by setting its group.id in the properties file for the consumer. To use the subscribe or commit methods provided by the KafkaConsumer API, you must assign the consumer to a consumer group by setting the group.id property. If you don’t, an exception occurs when these methods are called.

When assigned to a group, the partitions of all the topics are divided among the consumers in the group. As new group members arrive and old members leave, the partitions are re-assigned so that each member receives a proportional share of the partitions. This is known as rebalancing the group.

One of the brokers is designated as the group’s coordinator and is responsible for managing the members of the group as well as their partition assignments. The coordinator of each group is chosen from the leaders of the internal offsets topic, __consumer_offsets, which is used to store committed offsets. Basically, the group’s ID is hashed to one of the partitions for this topic, and the leader of that partition is selected as the coordinator. In this way, management of consumer groups is divided roughly equally across all the brokers in the cluster, which allows the number of groups to scale by increasing the number of brokers.

When the consumer starts up, it finds the coordinator for its group and sends a request to join the group. The coordinator then begins a group rebalance so that the new member is assigned its fair share of the group’s partitions. Every rebalance results in a new generation of the group.

Each member in the group must send heartbeats to the coordinator in order to remain a member of the group. If no heartbeat is received before expiration of the configured session timeout, then the coordinator will kick the member out of the group and reassign its partitions to another member.

When a consumer commits offsets, the group coordinator validates that the consumer still owns the partitions it is committing for. This prevents consumers that have been fenced or whose partitions were reassigned from overwriting offsets for partitions they no longer own. The coordinator uses epoch validation for consumer groups using the consumer group protocol (group.protocol=consumer). The coordinator tracks an assignment epoch per partition rather than a single epoch per member, and uses those per-partition epochs to validate offset commits.

For a short video that describes consumer groups, group leaders, and group coordinators, watch:

Groups and rebalance protocols

Recall that rebalancing is the process of assigning (or reassigning) Kafka topic partitions to the members of a consumer group. Kafka supports two rebalance protocols that determine how rebalancing happens within a consumer group. These protocols are:

  • classic, which was the only protocol before Kafka version 4.0

  • consumer, a newer protocol which has General Availability (GA) in the Kafka 4.0 version (Confluent Platform 8.0)

Warning

The classic rebalance protocol will be replaced by the consumer protocol in upcoming releases of Kafka clients.

Kafka clients 4.3 and later (Confluent Platform 8.3 and later)

The consumer protocol is recommended. The classic protocol issues a warning that deprecation will occur in Kafka clients 5.0.

Kafka clients 5.0 and later

The consumer protocol is the default. The classic protocol remains available but is deprecated.

Kafka clients 6.0 and later

There is no client-level support for the classic protocol.

For more information, see KIP-1274: Deprecate and remove support for Classic rebalance protocol in KafkaConsumer.

Overview of the rebalance protocols

Confluent Platform supports both the classic and the consumer rebalance protocol, so your consumer clients can use either protocol. The consumer rebalance protocol is preferred.

The consumer rebalance protocol improves consumer group scalability by removing the group-wide synchronization barrier, making rebalancing truly incremental. Removing this barrier also enhances stability by preserving existing partition assignments as much as possible during rebalances. The consumer rebalance protocol also reduces rebalance times and simplifies consumers. By offloading or simplifying consumer-side responsibilities, the entire rebalance is streamlined and less prone to delays caused by complex client-side operations.

Regardless of the rebalance protocol in use, the coordinator begins the group formation process by validating and finalizing the partition assignments. The coordinator computes assignments based on the topic subscriptions and assignment preferences communicated by the members (in the consumer protocol) or by the group leader (in the classic protocol).

The following table shows the differences between the classic and consumer rebalance protocols:

Group behavior

Classic protocol

Consumer protocol

use case

Good for architectures that need to continue support for the classic protocol.

Ideal for dynamic groups, such as deployments that scale up and down frequently.

group leader

One member performs assignment on behalf of the group.

Not used. No leader election occurs.

coordinator role

Assignments computed by the group leader (a consumer) and applied by the broker coordinator.

Broker coordinator computes assignments but does not receive one global plan.

rebalance type

Eager or cooperative: assignor dependent. In some circumstances, all consumers can pause and revoke all partitions.

Incremental reassignment with minimal disruption.

join group flow

Triggered on join or leave; reassigns all partitions.

Incremental assignment; a consumer joins a group by sending a heartbeat request; fewer disruptions.

assignment

Handled by one group leader among the consumers.

Broker coordinator receives consumer subscriptions, validates them, computes, and then orchestrates assignment.

Enable the consumer rebalance protocol on brokers

In Confluent Platform, you are responsible for enabling the consumer rebalance protocol on your own brokers. Two broker-side settings control availability of the protocol:

  • The group.coordinator.rebalance.protocols broker configuration must include consumer in its list of enabled rebalance protocols on Confluent Platform versions earlier than 8.0. This configuration is deprecated and will be removed with the Confluent Platform version aligned to Kafka 5.0.

  • The group.version feature flag must be set to a version that supports the protocol. On new Confluent Platform 8.0 or later clusters, group.version=1 is enabled by default. On clusters upgraded from an earlier release, upgrade the feature explicitly:

    ${CONFLUENT_HOME}/bin/kafka-features --bootstrap-server localhost:9092 upgrade --feature group.version=1

    To disable the feature, downgrade it:

    ${CONFLUENT_HOME}/bin/kafka-features --bootstrap-server localhost:9092 downgrade --feature group.version=0

For more information about broker-side settings that tune consumer groups using the consumer rebalance protocol, such as group.consumer.session.timeout.ms, group.consumer.heartbeat.interval.ms, and group.consumer.assignors, see Kafka Broker Configurations.

How a group’s rebalance protocol is determined

A Kafka consumer group is implicitly created when the first consumer with a specific group.id starts and attempts to subscribe to a topic. No explicit “create group” command or administrative action is required.

The first consumer to join a group determines the rebalance protocol for that group. If this initial consumer has the configuration group.protocol=consumer set, the group uses the consumer rebalance protocol. Otherwise, the group defaults to the classic rebalance protocol.

For example, you can change a group’s rebalance protocol by shutting down all of its consumers and bringing them back up. The following is a step-by-step explanation of how a consumer group’s protocol can change from classic to the consumer rebalance protocol:

  1. Initially, in this scenario, the consumer group is operating with the classic protocol. The broker’s group coordinator stores metadata about this group, including the protocol it’s currently using (classic).

  2. Shut down all the consumer instances belonging to this group. At this point, the group becomes empty from the broker’s perspective. No active members sending heartbeat requests.

    Even though the group is empty, the broker retains the metadata associated with that group, including the last known protocol it was using (classic in this scenario). The group isn’t entirely “forgotten.”

  3. The first consumer instance you bring back up is configured with group.protocol=consumer. This consumer sends a request to the coordinator to join a group. This request explicitly specifies that it wants to use the consumer protocol.

  4. Because the group is currently empty, the broker accepts this new member and updates the stored metadata for that consumer group. The broker now records that this group should use the consumer rebalance protocol.

As other consumer instances (also configured with group.protocol=consumer) are brought back online and join using the same group.id, the broker sees that the group is now configured to use the consumer protocol and allows them to join using that protocol.

A consumer using the classic protocol that attempts to join isn’t rejected unless it uses an incompatible protocol, for example, Connect. Typically, the join fails with an InconsistentGroupProtocolException exception.

Upgrade or switch consumer protocols

You can upgrade or switch consumer groups between the classic and consumer protocols. If you want to migrate or upgrade your clients to the consumer rebalance protocol, your choice is between a rolling upgrade or an empty group restart. When choosing between the two approaches, take these aspects of your consumers into consideration:

  • Client version compatibility. Check and ensure that your consumer libraries support the Kafka 4.0 version and the group.protocol configuration. For the consumer rebalance protocol to be effective, all consumers in the group should use it.

  • Broker support. The broker’s group.version feature flag must be set to a version that supports the desired protocol, and group.coordinator.rebalance.protocols must include consumer. For more information, see Enable the consumer rebalance protocol on brokers.

Rolling upgrades are the preferred method for production environments because they minimize downtime. If your environment can tolerate some downtime, you can do an empty group restart.

You can also switch back to the classic rebalance protocol from the consumer rebalance protocol. In this case, you would do a rolling upgrade, but switch from group.protocol=consumer back to the default, either by removing the setting or by setting it explicitly to group.protocol=classic. If you choose this path, make sure that you have a good understanding of the potential feature loss and use careful coordination among your team.

Note

If a consumer attempts to join a group operating with an incompatible protocol, the consumer receives an InconsistentGroupProtocolException.

Before you migrate to the consumer rebalance protocol

Check your consumers and adjust them as necessary before migrating or upgrading to the consumer rebalance protocol. A consumer uses the consumer rebalance protocol under the following conditions:

  • The consumer uses a client version that supports the consumer protocol.

  • The consumer configuration explicitly sets the group.protocol=consumer property.

Ensure that classic legacy configurations are removed from your consumer properties. Several configurations and two APIs from the classic rebalance protocol are no longer applicable in the consumer rebalance protocol:

  • The heartbeat.interval.ms property on the consumer side is replaced by the server-side group.consumer.heartbeat.interval.ms in the consumer rebalance protocol.

  • The session.timeout.ms property on the consumer side is replaced by the server-side group.consumer.session.timeout.ms in the consumer rebalance protocol.

  • The partition.assignment.strategy property on classic consumers is replaced in the consumer rebalance protocol by the server-side group.consumer.assignors on the broker and the group.remote.assignor on the consumer.

  • The enforceRebalance(String) and enforceRebalance() APIs are no longer supported with consumers using the consumer rebalance protocol.

If your consumer uses any legacy properties or methods, they are either ignored or result in errors when you use them with a consumer where group.protocol=consumer.

The consumer rebalance protocol also includes subscribe(SubscriptionPattern) and subscribe(SubscriptionPattern,ConsumerRebalanceListener) methods. These methods allow consumers to subscribe to a regular expression. With these methods, the regular expression uses the RE2J format and is evaluated on the server side.

Tip

For information about using the command line to manage your consumer groups, see Kafka consumer group tool later in this page.

Do a rolling deployment

When you upgrade or switch a consumer group to use the group.protocol=consumer configuration with a rolling deployment, a group can temporarily exist in a transitional state where some consumers are using the consumer rebalance protocol and others the classic protocol. While Kafka attempts to interoperate between these protocols during this phase, this is not a recommended long-term operating model and can lead to issues and limitations such as the following:

  • The broker must maintain compatibility with both protocols, which can introduce overhead and prevent the group from fully leveraging the optimizations of the consumer protocol.

  • Managing a group with mixed protocols can increase the complexity of the rebalance process and potentially lead to less predictable behavior, especially in edge cases or during rapid membership changes.

  • The differences in how the protocols handle assignment can lead to temporary imbalances in partition distribution among consumers using different protocols.

  • Monitoring and troubleshooting the rebalance process and group health is more difficult when different consumers are operating under different protocol rules.

Follow this procedure to do a rolling upgrade:

  1. Ensure that all consumers in the group are using a client version that supports the consumer rebalance protocol.

  2. Remove any classic configurations from the consumer properties.

  3. Deploy new versions of your consumer applications configured with group.protocol=consumer, one at a time.

  4. Restart each consumer instance after the configuration change.

  5. As the first consumer protocol consumer joins an existing classic group with a compatible assignor, the group’s metadata on the broker is updated.

    You can use the kafka-consumer-groups command-line tool to check the migration progress. For information about how to use this tool, see Kafka consumer group tool later in this page.

  6. Continue rolling out the new configuration to all consumers in the group.

  7. After you upgrade all consumers, the group fully uses the consumer protocol.

Do an empty group restart

This method is simpler than a rolling restart but involves more downtime.

  1. Shut down all consumer instances in the group, making the group empty.

  2. Ensure that all consumers in the group are using a client version that supports the consumer rebalance protocol.

  3. Remove any classic configurations from the consumer properties.

  4. Bring all consumer instances back online with the group.protocol=consumer configuration.

  5. The first joining consumer with the new protocol sets the group’s protocol for subsequent members.

Tip

Kafka Streams applications use a separate, dedicated rebalance protocol. For more information, see Streams Rebalance Protocol for Kafka Streams in Confluent Platform.

Offset management

After the consumer receives its assignment from the coordinator, it must determine the initial position for each assigned partition. When the group is first created, before any messages have been consumed, the position is set according to a configurable offset reset policy (auto.offset.reset). Typically, consumption starts either at the earliest offset or the latest offset.

As a consumer in the group reads messages from the partitions assigned by the coordinator, it must commit the offsets corresponding to the messages it has read. If the consumer crashes or is shut down, its partitions will be re-assigned to another member, which will begin consumption from the last committed offset of each partition. If the consumer crashes before any offset has been committed, then the consumer which takes over its partitions will use the reset policy.

The offset commit policy is crucial to providing the message delivery guarantees needed by your application. By default, the consumer is configured to use an automatic commit policy, which triggers a commit on a periodic interval. The consumer also supports a commit API which can be used for manual offset management. Correct offset management is crucial because it affects delivery semantics.

By default, the consumer is configured to auto-commit offsets. The auto.commit.offset.interval property sets the upper time bound of the commit interval. Using auto-commit offsets can give you “at-least-once” delivery, but you must consume all data returned from a ConsumerRecords<K, V> poll(Duration timeout) call before any subsequent poll calls, or before closing the consumer.

To explain further; when auto-commit is enabled, every time the poll method is called and data is fetched, the consumer is ready to automatically commit the offsets of messages that have been returned by the poll. If the processing of these messages is not completed before the next auto-commit interval, there’s a risk of losing the message’s progress if the consumer crashes or is otherwise restarted. In this case, when the consumer restarts, it will begin consuming from the last committed offset. When this happens, the last committed position can be as old as the auto-commit interval. Any messages that have arrived since the last commit are read again.

If you want to reduce the window for duplicates, you can reduce the auto-commit interval, but some users may want even finer control over offsets. The consumer therefore supports a commit API which gives you full control over offsets. Note that when you use the commit API directly, you should first disable auto-commit in the configuration by setting the enable.auto.commit property to false.

Each call to the commit API results in an offset commit request being sent to the broker. Using the synchronous API, the consumer is blocked until that request returns successfully. This may reduce overall throughput since the consumer might otherwise be able to process records while that commit is pending.

One way to deal with this is to increase the amount of data that is returned when polling. The consumer has a configuration setting fetch.min.bytes which controls how much data is returned in each fetch. The broker will hold on to the fetch until enough data is available (or fetch.max.wait.ms expires). The tradeoff, however, is that this also increases the amount of duplicates that have to be dealt with in a worst-case failure.

A second option is to use asynchronous commits. Instead of waiting for the request to complete, the consumer can send the request and return immediately by using asynchronous commits.

So if it helps performance, why not always use asynchronous commits? The main reason is that the consumer does not retry the request if the commit fails. This is something that committing synchronously gives you for free; it will will retry indefinitely until the commit succeeds or an unrecoverable error is encountered. The problem with asynchronous commits is dealing with commit ordering. By the time the consumer finds out that a commit has failed, you may already have processed the next batch of messages and even sent the next commit. In this case, a retry of the old commit could cause duplicate consumption.

Instead of complicating the consumer internals to try and handle this problem in a sane way, the API gives you a callback which is invoked when the commit either succeeds or fails. If you like, you can use this callback to retry the commit, but you will have to deal with the same reordering problem.

Offset commit failures are merely annoying if the following commits succeed since they won’t actually result in duplicate reads. However, if the last commit fails before a rebalance occurs or before the consumer is shut down, then offsets will be reset to the last commit and you will likely see duplicates. A common pattern is therefore to combine async commits in the poll loop with sync commits on rebalances or shut down. Committing on close is straightforward, but you need a way to hook into rebalances.

Each rebalance has two phases: partition revocation and partition assignment. The revocation method is always called before a rebalance and is the last chance to commit offsets before the partitions are re-assigned. The assignment method is always called after the rebalance and can be used to set the initial position of the assigned partitions. In this case, the revocation hook is used to commit the current offsets synchronously.

In general, asynchronous commits should be considered less safe than synchronous commits. Consecutive commit failures before a crash will result in increased duplicate processing. You can mitigate this danger by adding logic to handle commit failures in the callback or by mixing occasional synchronous commits, but you shouldn’t add too much complexity unless testing shows it is necessary. If you need more reliability, synchronous commits are there for you, and you can still scale up by increasing the number of topic partitions and the number of consumers in the group. But if you just want to maximize throughput and you’re willing to accept some increase in the number of duplicates, then asynchronous commits may be a good option.

A somewhat obvious point, but one that’s worth making is that asynchronous commits only make sense for “at least once” message delivery. To get “at most once,” you need to know if the commit succeeded before consuming the message. This implies a synchronous commit unless you have the ability to “unread” a message after you find that the commit failed.

In the examples, we show several detailed examples of the commit API and discuss the tradeoffs in terms of performance and reliability.

When writing to an external system, the consumer’s position must be coordinated with what is stored as output. That is why the consumer stores its offset in the same place as its output. For example, a connector populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and offsets are both updated, or neither is. A similar pattern is followed for many other data systems that require these stronger semantics, and for which the messages do not have a primary key to allow for deduplication.

This is how Kafka supports exactly-once processing in Kafka Streams, and the transactional producer or consumer can be used generally to provide exactly-once delivery when transferring and processing data between Kafka topics. Otherwise, Kafka guarantees at-least-once delivery by default, and you can implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.

Tip

Consumers can fetch and consume messages from out-of-sync follower replicas if using a fetch-from-follower configuration. For more information, see Multi-Region Clusters.

Kafka consumer configuration

The full list of configuration settings are available in Kafka Consumer Configurations. Several of the key configuration settings and how they affect the consumer’s behavior are highlighted below.

Core configuration

Following are some important consumer properties:

  • bootstrap.servers: You are required to set this property so that the consumer can find the Kafka cluster.

  • client.id: Optional, but you should set this property to easily correlate requests on the broker with the client instance which made it. Typically, all consumers within the same group will share the same client ID in order to enforce client quotas.

Group configuration

The following properties apply to consumer groups.

  • group.id: Optional but you should always configure a group ID unless you are using the simple assignment API and you don’t need to store offsets in Kafka.

  • session.timeout.ms: Control the session timeout by overriding this value. The default is 10 seconds in the C/C++ and Java clients, but you can increase the time to avoid excessive rebalancing, for example due to poor network connectivity or long GC pauses. The main drawback to using a larger session timeout is that it will take longer for the coordinator to detect when a consumer instance has crashed, which means it will also take longer for another consumer in the group to take over its partitions. For normal shutdowns, however, the consumer sends an explicit request to the coordinator to leave the group which triggers an immediate rebalance.

  • heartbeat.interval.ms: This controls how often the consumer will send heartbeats to the coordinator. It is also the way that the consumer detects when a rebalance is needed, so a lower heartbeat interval will generally mean faster rebalancing. The default setting is three seconds. For larger groups, it may be wise to increase this setting.

  • max.poll.interval.ms: This property specifies the maximum time allowed time between calls to the consumers poll method (Consume method in .NET) before the consumer process is assumed to have failed. The default is 300 seconds and can be safely increased if your application requires more time to process messages. If you are using the Java consumer, you can also adjust max.poll.records to tune the number of records that are handled on every loop iteration.

Offset management configuration

There are two main settings for offset management; whether auto-commit is enabled and the offset reset policy.

  • enable.auto.commit: This setting enables auto-commit (the default), which means the consumer automatically commit offsets periodically at the interval set by auto.commit.interval.ms. The default interval is 5 seconds.

  • auto.offset.reset: Defines the behavior of the consumer when there is no committed position (which occurs when the group is first initialized) or when an offset is out of range. There are several ways to set the offset. You can choose either to reset the position to the earliest offset or the latest offset (the default). You can also reset to a configured duration from the current timestamp.

Partition assignment configuration

partition.assignment.strategy sets the partition assignment strategy for a consumer, meaning how partition ownership is distributed between consumer instances when group management is used. All consumers in the same consumer group must have the same partition strategy.

The partition.assignment.strategy parameter is not supported when the consumer group protocol, group.protocol=consumer, is enabled. This configuration applies only when using the classic consumer group protocol, group.protocol=classic, which is the default.

Warning

The classic rebalance protocol will be replaced by the consumer protocol in upcoming releases of Kafka clients. For the deprecation timeline and migration guidance, see Groups and rebalance protocols.

partition.assignment.strategy accepts a comma-separated list of fully qualified class names that implement the PartitionAssignor interface. The list enables you to update the strategy for a group, while temporarily keeping the old one for consumers that have not transitioned to the new strategy yet.

When you configure Kafka consumers, the choice of assignment strategy is important and depends on the specific requirements for partition balancing, consumer group stability, and rebalance behavior. In most cases, the default (range assignment) works well, but for specific use cases, changing the assignment strategy can significantly impact performance and reliability.

Available options are:

  • Range Assignment (Default)

    • How it Works: The org.apache.kafka.clients.consumer.RangeAssignor works by evenly distributing partitions of each topic across the consumers in a consumer group. It sorts both the partitions and consumers. Partitions are assigned to consumers in chunks (ranges), aiming for an even distribution.

    • Advantages: It works well when partition count is higher than consumer count, providing a simple and efficient means of partition distribution.

    • Disadvantages: Can result in uneven load distribution if the number of partitions is not a multiple of the number of consumers.

  • Round Robin Assignment

    • How it Works: The org.apache.kafka.clients.consumer.RoundRobinAssignor distributes partitions across all consumers one by one in a round-robin fashion. It ensures a more even distribution of partitions across consumers, regardless of the number of partitions.

    • Advantages: Leads to a more balanced partition allocation across consumers, useful when handling a varying number of partitions or when partitions have significantly different sizes.

    • Disadvantages: May lead to more rebalances compared to Range Assignment in some scenarios.

  • Sticky Assignor

    • How it Works: The org.apache.kafka.clients.consumer.StickyAssignor aims to maintain a stable partition assignment while still balancing the partitions across consumers. It tries to keep the previously assigned partitions to a consumer as unchanged as possible if a rebalance occurs.

    • Advantages: It minimizes the number of partition reassignments across rebalances, reducing the potential for missed messages or duplicated processing.

    • Disadvantages: While it offers stability, it might not always result in the most balanced partition distribution if the cluster or consumer group changes frequently.

  • Cooperative Sticky Assignor

    • How it Works: An evolution of the StickyAssignor, the org.apache.kafka.clients.consumer.CooperativeStickyAssignor enables more incremental rebalancing, which can reduce the latency and resources required during the rebalance process.

    • Advantages: It supports more granular changes to the consumer group memberships or to the partitions themselves, making rebalances less impactful. Note that this assignor reduces rebalance impact, not the frequency of rebalances. For consumer groups where all members subscribe to the same set of topics, the org.apache.kafka.clients.consumer.ConstrainedCooperativeStickyAssignor provides the same benefits and is optimized for this common scenario.

    • Disadvantages: Not all consumers or versions of Kafka support this assignor; requires careful management to ensure compatibility across the consumer group.

Upgrade to CooperativeStickyAssignor

To migrate an existing consumer group from an eager rebalance assignor (for example, RangeAssignor or StickyAssignor) to the cooperative protocol introduced by KIP-429: Kafka Consumer Incremental Rebalance Protocol, use a two-step rolling bounce to minimize disruption during the transition. After completing step 1 (configuring dual assignors), the consumer group operates in a mixed-mode period where members that still use an eager assignor continue to perform full-stop rebalances. After completing step 2 (switching to the cooperative assignor), rebalances become incremental and avoid full-stop rebalances.

Note

CooperativeStickyAssignor requires Kafka client version 2.4.0 or later. Ensure all consumers in your group meet this version requirement before beginning the upgrade.

  1. Prepare (dual list, keep current first).

    Set partition.assignment.strategy to include both your current assignor and CooperativeStickyAssignor, with the current assignor listed first. Perform a rolling restart so all members pick up the new list. The coordinator will continue to select the first commonly supported assignor (the existing one), so there is no behavior change yet.

    Example properties:

    partition.assignment.strategy=org.apache.kafka.clients.consumer.StickyAssignor,org.apache.kafka.clients.consumer.CooperativeStickyAssignor

    Fully qualified class names shown here refer to the Java consumer. For other client libraries, configure the equivalent assignor using the client’s configuration API.

    Example Java configuration:

    props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
          Arrays.asList(StickyAssignor.class.getName(),
                        CooperativeStickyAssignor.class.getName()));

  2. Switch (make cooperative first or only).

    After all members have been upgraded to client versions that support cooperative rebalancing, update the configuration to make CooperativeStickyAssignor the first (or only) value and roll the group again. The coordinator then selects cooperative rebalancing and the group transitions incrementally with minimal disruption.

    Example properties:

    partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

    Fully qualified class names shown here refer to the Java consumer. For other client libraries, configure the equivalent assignor using the client’s configuration API.

    Example Java configuration:

    props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
          Collections.singletonList(CooperativeStickyAssignor.class.getName()));

    Verify the switch: after the rollout, check consumer application logs for messages indicating the selected partition assignor (for example, “Using partition assignor class …CooperativeStickyAssignor”).

Notes

  • All members in the group must advertise a compatible, ordered list of assignors. The coordinator selects the first assignor that is present in every member’s list.

  • Do not switch to listing only CooperativeStickyAssignor until all members run client versions that support it (2.4.0 or later); otherwise the group fails to find a common assignor.

  • Static membership (group.instance.id) further reduces movement during restarts and pairs well with cooperative rebalancing. In Kubernetes environments, use StatefulSets to provide stable pod identities that can serve as static member IDs.

  • If using manual offset commits (enable.auto.commit=false), handle RebalanceInProgressException by calling poll() in the next loop iteration to complete the rebalancing process.

  • For background and the detailed upgrade rationale, see KIP-429: Kafka Consumer Incremental Rebalance Protocol (see the “Compatibility and Upgrade Path” section).

Message handling

While the Java consumer does all IO and processing in the foreground thread, librdkafka-based clients (C/C++, Python, Go and C#) use a background thread. The main consequence of this is that polling is totally safe when used from multiple threads. You can use this to parallelize message handling in multiple threads. From a high level, poll is taking messages off of a queue which is filled in the background.

Another consequence of using a background thread is that all heartbeats and rebalancing are executed in the background. The benefit of this is that you don’t need to worry about message handling causing the consumer to “miss” a rebalance. The drawback, however, is that the background thread will continue heart beating even if your message processor dies. If this happens, then the consumer will continue to hold on to its partitions and the read lag will continue to build until the process is shut down.

Although the clients have taken different approaches internally, they are not as far apart as they seem. To provide the same abstraction in the Java client, you could place a queue in between the poll loop and the message processors. The poll loop would fill the queue and the processors would pull messages off of it.

Kafka consumer group tool

Kafka includes the kafka-consumer-groups command-line utility to view and manage consumer groups, which is also provided with Confluent Platform. Find the tool in the bin folder under your installation directory.

You can also use the Confluent CLI to complete some of these tasks. For more information, see the Confluent CLI reference.

List consumer groups

You can get a list of the active groups in the cluster using the kafka-consumer-groups tool. On a large cluster, this may take a while since it collects the list by inspecting each broker in the cluster.

bin/kafka-consumer-groups --bootstrap-server host:9092 --list

Your output will be a list of list of all consumer groups for the cluster, including consumers for internal use. The output might resemble:

_confluent-controlcenter-7-6-0-1
ConfluentTelemetryReporterSampler--4418883999569981189
test-1234
_confluent-controlcenter-7-6-0-lastProduceTimeConsumer
_confluent-controlcenter-7-6-0-1-command

To use the Confluent CLI for this task, see confluent kafka consumer group list.

Describe groups

The kafka-consumer-groups tool can also be used to collect information on a current group. For example, to see the current assignments for the test-1234 group, you could use the following command:

bin/kafka-consumer-groups --bootstrap-server host:9092 --describe --group test-1234

The output from this command will list the clients, topics, partitions and more for that group. The output might resemble:

GROUP      TOPIC        PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID        HOST            CLIENT-ID
test-1234  test-metrics 5          258420          258519          -               test-client-1234    /127.0.0.1     test-client
test-1234  test-metrics 10         257002          257097          -               test-client-1234    /127.0.0.1     test-client
test-1234  test-metrics 4          259580          259660          -               test-client-1234    /127.0.0.1     test-client
test-1234  test-metrics 7          254004          254131          -               test-client-1234    /127.0.0.1     test-client

If you happen to invoke this while a rebalance is in progress, the command will report an error. Retry again and you should see the assignments for all the members in the current generation.

To use the Confluent CLI for this task, see confluent kafka consumer group describe.

Reset offsets

You can also use this tool to reset the consumer offset in scenarios where a consumer is stalled or significantly lagging. Make sure you deactivate the consumer before you reset its offset. You have many options for changing the offset. For example, you can reset offsets by shifting forward or backward with shift-by or reset them to the beginning with --to-earliest. For all the options, see the kafka-consumer-group tool usage details.

To reset the offsets back by 20 positions, use the following command:

bin/kafka-consumer-groups.sh --bootstrap-server host:9092 --group test-1234 --reset-offsets --shift-by -20 --topic test-metrics -execute --group test-1234

The output will contain the group, topic and partitions, and the new offset for each:

GROUP             TOPIC                          PARTITION  NEW-OFFSET
test-1234         test-metrics                   5          258400
test-1234         test-metrics                   10         256082
test-1234         test-metrics                   4          259560
test-1234         test-metrics                   7          254984

Consumer examples

Confluent provides a number of resources to help you get started with Kafka consumers.

Related content