Share Consumers for Confluent Platform

Share consumers form share groups, an alternative to traditional consumer groups that enables concurrent processing from the same partition.

Traditional consumer groups use a distributed processing model where consumer members collectively read messages from an Apache Kafka® topic. In the traditional model, each topic partition is assigned to only one group member at a time. In contrast, share groups and their share consumer members use a concurrent processing model. Share group members simultaneously process different messages from the same partition, effectively treating it more like a message queue.

Important

Share groups are only available for the Kafka Java clients.

Key capabilities and differences from consumer groups

Share groups enable Kafka share consumers to cooperatively consume and process records from topics. This contrasts with traditional consumer groups where each partition is assigned to only one consumer within the group.

The fundamental differences between a share group and a consumer group are:

Cooperative consumption

Consumers within a share group cooperatively consume records, and partitions may be assigned to multiple consumers. This allows for increased concurrency beyond the number of topic partitions.

Scalability beyond partitions

The number of consumers in a share group can exceed the number of partitions in a topic. This is a significant advantage over traditional consumer groups where the maximum parallelism is limited by the number of partitions.

Individual record acknowledgement

Records are acknowledged individually, though the system is optimized for batch processing to improve efficiency.

Delivery attempt counting

Delivery attempts to consumers in a share group are counted, which enables automated handling of unprocessable records.

When a share consumer in a share group fetches records, it receives available records from any of the topic-partitions matching its subscriptions. Records are acquired for delivery to this share consumer with a time-limited acquisition lock. While a record is acquired, it is unavailable to other share consumers. The default lock duration is 30 seconds, controlled by the share.record.lock.duration.ms group configuration parameter. The lock is released automatically once its duration elapses, making the record available to another share consumer.

A share consumer holding the lock can handle the record in the following ways:

• Acknowledge successful processing of the record.

• Release the record, making it available for another delivery attempt.

• Renew the lock, enabling long running processing in the share consumer.

• Reject the record, indicating it’s unprocessable and preventing further delivery attempts for that record.

• Do nothing, in which case the lock is automatically released when its duration expires.

The Kafka cluster limits the number of records acquired for consumers for each topic-partition within a share group. Once this limit is reached, fetching operations will temporarily yield no further records until the number of acquired records decreases (as locks naturally time out). This limit is controlled by the broker configuration property group.share.partition.max.record.locks. By limiting the duration of the acquisition lock and automatically releasing the locks, the broker ensures delivery progresses even in the presence of share consumer failures.

Configuration for client developers

To create a share consumer, create an instance of KafkaShareConsumer in a Java application. When developing a client application that utilizes share groups, there are share consumer-level configuration properties and group-level configuration properties you can define.

You define the following share consumer-level properties in your consumer code:

share.acknowledgement.mode

• Description: Controls the acknowledgement mode for a share consumer.

• Type: string

• Default: implicit

• Valid Values: [implicit, explicit]

• Importance: medium

If your KafkaShareConsumer configuration sets share.acknowledgement.mode=explicit then, every record returned from poll must be explicitly acknowledged with KafkaShareConsumer.acknowledge.

If you do not set the share.acknowledgement.mode configuration or if you set share.acknowledgement.mode=implicit, then implicit mode is used, which automatically acknowledges records.

share.acquire.mode

• Description: Controls how a share consumer obtains records.

• Type: string

• Default: batch_optimized

• Valid Values: [batch_optimized, record_limit]

• Importance: medium

The behavior of the max.poll.records client configuration will change depending on the selected acquire mode.

If set to record_limit, the number of records returned in each poll() will not exceed the value of max.poll.records.

If set to batch_optimized, the number of records returned in each poll() call may exceed max.poll.records to align with batch boundaries for optimization.

You can define group-level properties using the Confluent CLI or the REST Proxy.

You can also define these group-level configuration properties using the kafka-configs tool.

share.auto.offset.reset

• Description: The strategy to initialize the share-partition start offset.

• Type: string

• Default: latest

• Valid Values: [latest, earliest, by_duration:<duration>]

• Importance: medium

share.heartbeat.interval.ms

• Description: The heartbeat interval given to the members of a share group.

• Type: int

• Default: 5000 (5 seconds)

• Valid Values: [1,…]

• Importance: medium

If not specified, uses the broker’s group.share.heartbeat.interval.ms configuration. If specified, the minimum is limited by the broker’s group.share.min.heartbeat.interval.ms configuration, and the maximum is limited by the broker’s group.share.max.heartbeat.interval.ms configuration.

share.isolation.level

• Description: Controls how to read records written transactionally.

• Type: string

• Default: read_uncommitted

• Valid Values: [read_committed, read_uncommitted]

• Importance: medium

If set to read_committed, the share group only delivers transactional records which were committed. If set to read_uncommitted, the share group returns all messages, even transactional messages which were aborted. Non-transactional records are returned unconditionally in either mode.

share.record.lock.duration.ms

• Description: The record acquisition lock duration in milliseconds for share groups.

• Type: int

• Default: 30000 (30 seconds)

• Valid Values: [1000,…]

• Importance: medium

If not specified, uses the broker’s group.share.record.lock.duration.ms configuration. If specified, the minimum is limited by the broker’s group.share.min.record.lock.duration.ms configuration, and the maximum is limited by the broker’s group.share.max.record.lock.duration.ms configuration.

share.session.timeout.ms

• Description: The timeout to detect client failures when using the share group protocol.

• Type: int

• Default: 45000 (45 seconds)

• Valid Values: [1,…]

• Importance: medium

If not specified, uses the broker’s group.share.session.timeout.ms configuration. If specified, the minimum is limited by the broker’s group.share.min.session.timeout.ms configuration, and the maximum is limited by the broker’s group.share.max.session.timeout.ms configuration.

share.delivery.count.limit

• Description: Maximum delivery attempts before a record is skipped.

• Type: int

• Default: 5

• Valid Values: [2, …]

• Importance: medium

Controls how many times the broker will attempt to deliver a record to share consumers before skipping it. The value must fall within the bounds set by the broker’s group.share.min.delivery.count.limit and group.share.max.delivery.count.limit configurations.

Important

Production deployments should explicitly set this value based on workload characteristics rather than relying on the default. Defaults are permissive for experimentation but may not be suitable for production use.

share.partition.max.record.locks

• Description: Maximum concurrent record locks per partition.

• Type: int

• Default: 2,000

• Maximum: 10,000

• Valid Values: [100, …]

• Importance: medium

Limits the number of records that can be acquired (locked) simultaneously for consumers in this share group for each topic partition. After this limit is reached, fetch operations don’t yield any further records until the number of acquired records decreases. The value must fall within the bounds set by the broker’s group.share.min.partition.max.record.locks and group.share.max.partition.max.record.locks configurations.

Important

Production deployments should explicitly set this value based on workload characteristics rather than relying on the default. Defaults are permissive for experimentation but may not be suitable for production use.

share.renew.acknowledge.enable

• Description: Controls whether acknowledgment renewal is enabled.

• Type: boolean

• Default: true

• Valid Values: [true, false]

• Importance: medium

When set to false, attempts to renew acknowledgments return an INVALID_RECORD_STATE error. Use this setting to prevent long-running lock renewals in share consumers.

errors.deadletterqueue.topic.name

• Description: The name of the dead-letter queue (DLQ) topic assigned to the share group. The broker writes records that exhaust their delivery count limit or are explicitly rejected to this topic. If the string is empty, no DLQ is configured.

• Type: string

• Default: "" (empty string)

• Valid Values: topic name that starts with the broker’s errors.deadletterqueue.topic.name.prefix value and doesn’t start with __

• Importance: medium

errors.deadletterqueue.copy.record.enable

• Description: When true, copies the original record’s key and value into the DLQ record. When false, only the context headers using __dlq.errors.* are written.

• Type: boolean

• Default: false

• Valid Values: [true, false]

• Importance: medium

Broker configuration for share groups

Administrators can configure broker-level settings that set bounds for per-group share group configurations. These are configured in the broker’s server.properties file or via dynamic configuration.

group.share.max.delivery.count.limit

• Description: Upper bound for per-group delivery retry limit.

• Type: int

• Default: 10

• Valid Values: [5, 25]

• Importance: medium

Sets the maximum value that share.delivery.count.limit can be configured to for any share group on this broker.

group.share.min.delivery.count.limit

• Description: Lower bound for per-group delivery retry limit.

• Type: int

• Default: 2

• Valid Values: [2, 5]

• Importance: medium

Sets the minimum value that share.delivery.count.limit can be configured to for any share group on this broker.

group.share.max.partition.max.record.locks

• Description: Upper bound for per-group concurrent record locks.

• Type: int

• Default: 4,000

• Valid Values: [2000, 10000]

• Importance: medium

Sets the maximum value that share.partition.max.record.locks can be configured to for any share group on this broker.

group.share.min.partition.max.record.locks

• Description: Lower bound for per-group concurrent record locks.

• Type: int

• Default: 100

• Valid Values: [100, 2000]

• Importance: medium

Sets the minimum value that share.partition.max.record.locks can be configured to for any share group on this broker.

The following broker properties control DLQ topics for share groups:

errors.deadletterqueue.auto.create.topics.enable

• Description: When true, the broker automatically creates a share group’s configured DLQ topic if it does not exist. The broker sets errors.deadletterqueue.group.enable=true on the auto-created topic.

• Type: boolean

• Default: false

• Valid Values: [true, false]

• Importance: medium

errors.deadletterqueue.topic.name.prefix

• Description: Required naming prefix for any topic used as a DLQ. Set to an empty string using "" to remove the restriction.

• Type: string

• Default: "dlq."

• Valid Values: any string

• Importance: medium

For details about setting up DLQ, see Configure dead-letter queues for share groups.

Manage share groups

Administrators can use the kafka-share-groups tool to list, describe, reset, and delete share groups. Only share groups without any active members can be deleted. For information on using this tool, see the kafka-share-groups tool usage details. If you have Confluent Platform installed, find all of the command-line tools in the bin directory of your Confluent Platform installation.

List all share groups:

bin/kafka-share-groups --bootstrap-server localhost:9092 --list

View current start offset and lag of a share group:

bin/kafka-share-groups --bootstrap-server localhost:9092 --describe --group my-share-group

This command produces output like the following:

GROUP           TOPIC           PARTITION  START-OFFSET  LAG
my-share-group  topic1          0          4             0

The start offset is the earliest offset for in-flight records being evaluated for delivery to share consumers. Some records after the start offset may already have completed delivery.

Note

The admin client needs DESCRIBE access to all the topics used in the group.

Describe active members in a share group:

bin/kafka-share-groups --bootstrap-server localhost:9092 --describe --group my-share-group --members

Reset offsets of a share group to the latest offset:

bin/kafka-share-groups --bootstrap-server localhost:9092 --reset-offsets --group my-share-group --topic topic1 --to-latest --execute

This command produces output like the following:

GROUP           TOPIC           PARTITION  NEW-OFFSET
my-share-group  topic1          0          10

You can also use --to-earliest or --to-datetime <String: datetime> (format: YYYY-MM-DDThh:mm:ss.sss) to reset to different offsets. Use --dry-run instead of --execute to preview which offsets would be reset.

Delete offsets of individual topics in a share group:

bin/kafka-share-groups --bootstrap-server localhost:9092 --delete-offsets --group my-share-group --topic topic1

Delete one or more share groups:

bin/kafka-share-groups --bootstrap-server localhost:9092 --delete --group my-share-group

Configure dead-letter queues for share groups

A dead-letter queue (DLQ) is a topic that receives records a share group cannot process. The Kafka broker writes a record to the DLQ when either of these conditions occurs:

• The record’s delivery count reaches share.delivery.count.limit,

  which is set at a default of 5.

• A share consumer explicitly rejects the record with AcknowledgeType.REJECT.

The DLQ feature doesn’t require any consumer code changes. Any Kafka Java Client version 4.2 and later works with DLQ automatically.

Before you begin

DLQ activates only after share.version=2 is finalized on every broker cluster-wide. Until then, including during a rolling upgrade, records that exhaust their delivery count are archived without a DLQ write.

To check the finalized feature level and upgrade, use the kafka-features tool with these commands:

bin/kafka-features --bootstrap-server localhost:9092 describe
bin/kafka-features --bootstrap-server localhost:9092 upgrade --feature share.version=2

Set up a dead-letter queue

1. Optionally, define the prefix for DLQ topic names.

   By default, DLQ topic names must begin with dlq.. To use a different prefix, set it before creating the DLQ topic with errors.deadletterqueue.topic.name.prefix with the kafka-configs tool:

   bin/kafka-configs --bootstrap-server localhost:9092 \
     --entity-type brokers \
     --entity-default \
     --alter --add-config errors.deadletterqueue.topic.name.prefix=deadletter.
   

   The examples in the following steps use the default dlq. prefix. If you change the prefix, replace dlq. with your new prefix.

2. Create and configure the DLQ topic.

   Provide the DLQ topic using one of these methods:

   Auto-creation

   Set errors.deadletterqueue.auto.create.topics.enable=true on the cluster using the kafka-configs tool. The broker creates the DLQ topic and automatically sets errors.deadletterqueue.group.enable=true for the DLQ topic:

   bin/kafka-configs --bootstrap-server localhost:9092 \
     --entity-type brokers \
     --entity-default \
     --alter --add-config errors.deadletterqueue.auto.create.topics.enable=true
   

   Manual creation

   Create the new topic manually with the kafka-topics tool. Ensure the topic name starts with the errors.deadletterqueue.topic.name.prefix:

   bin/kafka-topics --bootstrap-server localhost:9092 \
     --create --topic dlq.my-topic
   
   bin/kafka-configs --bootstrap-server localhost:9092 \
     --entity-type topics \
     --entity-name dlq.my-topic \
     --alter --add-config errors.deadletterqueue.group.enable=true
   

   For more information about the errors.deadletterqueue.group.enable topic-level configuration, see the Topic configuration reference.

3. Assign the DLQ topic to the share group.

   bin/kafka-configs --bootstrap-server localhost:9092 \
     --entity-type groups \
     --entity-name my-share-group \
     --alter --add-config errors.deadletterqueue.topic.name=dlq.my-topic
   

4. Optionally, enable full-record copy.

   By default, only context headers are written to the DLQ record. You can also copy the original record’s key and value:

   bin/kafka-configs --bootstrap-server localhost:9092 \
     --entity-type groups \
     --entity-name my-share-group \
     --alter --add-config errors.deadletterqueue.copy.record.enable=true
   

DLQ record structure

When the broker writes a record to the DLQ topic, it adds the following context headers using the __dlq.errors. prefix to avoid collisions with application headers:

__dlq.errors.topic

The originating topic.

__dlq.errors.partition

The source partition.

__dlq.errors.offset

The source offset.

__dlq.errors.group

The share group ID.

__dlq.errors.delivery.count

The delivery attempt count at the time the record was archived.

__dlq.errors.message

The failure reason.

If errors.deadletterqueue.copy.record.enable is set to the default false, only the preceding context headers are written. When errors.deadletterqueue.copy.record.enable=true, the DLQ record also includes the original record’s key and value.

Limitations

Confluent Platform doesn’t support the following DLQ capabilities:

Re-queueing

No built-in tooling moves a record from the DLQ back to the original topic for redelivery. To re-queue records, write an application that consumes from the DLQ topic and produces to the original topic.

Compression

The broker does not compress records written to the DLQ.

Initialize share group offsets from an offset or a file

In addition to the built-in --to-earliest, --to-latest, and --to-datetime, you can initialize a share group at a specific offset, or at a set of offsets read from a CSV file. Initializing from a file lets you restore a share group to offsets you captured earlier. You can also move a share group to another cluster whose topics keep the same offsets, such as mirror topics created with Cluster Linking.

The following options require Kafka 4.4 or later, and the share group must have no active members:

--to-offset <Long: offset> Initialize the share-partition start offset to a specific offset. If the offset you specify is earlier than the earliest available offset, the earliest offset is used instead. If it’s later than the latest offset, the latest offset is used instead.

--to-current Initialize offsets to the current share-partition start offsets. Combine this option with --export to capture the offsets that a share group uses now.

--export Write the offsets in CSV format instead of a table. This option is only supported with --reset-offsets. To export the offsets without applying them, combine --reset-offsets with --dry-run. With --execute, the offsets are also applied.

--from-file <String: path to CSV file> Initialize offsets to the values defined in a CSV file.

Initialize a single share partition at a specific offset:

bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group <my_share_group> --topic <topic_name>:<partition> --to-offset <offset> --execute

Export the current offsets of a share group to a file:

bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group <my_share_group> --all-topics --to-current --export --dry-run > offsets.csv

Each line of the exported file holds a topic, a partition, and an offset:

topic1,0,1234
topic1,1,5678

Initialize a share group from the offsets in that file:

bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group <my_share_group> --from-file offsets.csv --execute

The tool also accepts a file that includes a group ID as its first column using group,topic,partition,offset as the format. In this case, the tool applies only the rows that match the group that you pass to --group. Use --dry-run in place of --execute to preview the offsets before you apply them.

Monitor share groups

Share partition lag

Share partition lag represents the number of records that have not yet been fully processed by consumers in a share group. Share partition lag is the primary metric used to automate the scaling of share consumer instances. Unlike standard consumer lag, share partition lag accounts for the non-contiguous state of records in these states:

• AVAILABLE: Not yet acquired by any consumer

• ACQUIRED: Currently assigned to a consumer with an active acquisition lock

• ACKNOWLEDGED: Processing complete

• ARCHIVED: Removed by compaction

Share partition lag is calculated as:

(highest offset) - (share-partition start offset) + 1 - (processed in-flight records)

To monitor lag effectively, you must distinguish between record states. Records in a Terminal state have been acknowledged or archived. These records no longer contribute to the lag calculation.

Users can retrieve share partition lag data using these APIs:

• Admin API using Admin.listShareGroupOffsets(), which returns SharePartitionOffsetInfo objects containing startOffset(), leaderEpoch(), and lag() values.

• JMX through kafka.server:type=tenant-metrics,group-protocol=share,partition={partitionid},group-id={sharegroupid},topic={topicname}, which exposes the lag value for the share partition.

Note

Share partition lag must be enabled for JMX monitoring to be available. Set the confluent.share.lag.calculator.enabled value to true. This property enables share lag monitoring, and is false by default.

Additionally, add org.apache.kafka.common.metrics.JmxReporter to the metric.reporters broker configuration to expose the metric in JMX.

Share group metrics

Metrics for share groups are emitted under the kafka.server:type=group-coordinator-metrics MBean. Relevant metrics include:

group-count,protocol={share}

Total number of groups per group type: Share.

share-acknowledgement

The total number of offsets acknowledged for share groups.

record-acknowledgement

The number of records acknowledged per acknowledgement type.

partition-load-time

The time taken to load the share partitions.

These metrics provide visibility into the health and performance of share groups on the Kafka cluster.

Dead-letter queue metrics

DLQ metrics are emitted under the kafka.server:type=ShareGroupMetrics,name={MetricName},group={group_id} MBean. Relevant metrics include:

DeadLetterQueueRecordCount

The total number of records written to the DLQ topic.

DeadLetterQueueTotalProduceRequestsPerSec

The rate of DLQ produce requests per second.

DeadLetterQueueFailedProduceRequestsPerSec

The rate of failed DLQ produce requests per second.

Troubleshooting share groups

INVALID_RECORD_STATE errors

If share consumers receive INVALID_RECORD_STATE errors, check the share.renew.acknowledge.enable configuration for the share group. This error is returned when acknowledgment renewal is attempted but share.renew.acknowledge.enable=false.

To resolve, either:

• Set share.renew.acknowledge.enable=true to allow acknowledgment renewal, or

• Modify the share consumer application to avoid renewing acknowledgments

Records not reaching the DLQ topic

If records that exhaust their delivery count or that a share consumer rejects are not written to the DLQ, check these conditions:

• The share.version feature level is lower than 2. DLQ support requires share.version=2 to be finalized cluster-wide. To fix this issue, run:

  kafka-features.sh --bootstrap-server localhost:9092 describe
  kafka-features.sh --bootstrap-server localhost:9092 upgrade --feature share.version=2
  

• The share group has no DLQ topic configured. Verify the errors.deadletterqueue.topic.name is set on the share group:

  bin/kafka-configs.sh --bootstrap-server localhost:9092 \
    --entity-type groups \
    --entity-name my-share-group \
    --describe
  

• The DLQ topic does not exist and auto-creation is off. If errors.deadletterqueue.auto.create.topics.enable=false, create the topic manually before configuring it as the DLQ.

• The DLQ topic does not have errors.deadletterqueue.group.enable=true. Set this configuration on the DLQ topic:

  bin/kafka-configs.sh --bootstrap-server localhost:9092 \
    --entity-type topics \
    --entity-name dlq.my-topic \
    --alter --add-config errors.deadletterqueue.group.enable=true
  

  Alternatively, enable auto-creation so the broker creates the topic with this configuration already set.

• The DLQ topic name does not start with errors.deadletterqueue.topic.name.prefix. Use a topic name that starts with the required prefix. Verify every existing DLQ topic name starts with the new prefix.

Production tuning recommendations

Share group defaults are permissive to help with experimentation and development. For production deployments:

• Explicitly set share.delivery.count.limit based on your workload’s retry requirements rather than using the broker default.

• Explicitly set share.partition.max.record.locks based on your expected concurrency and throughput requirements.

• Monitor share partition lag metrics to validate that configuration values are appropriate for your workload.

Kafka 4.1 version introduced share consumers and share groups. This section is an overview of share groups for client developers and highlights relevant configurations for broker administrators.

Manage access

You can control who has access to share groups by setting group and topic permissions using RBAC in the Confluent CLI.

Managing share group access requires the following permissions:

Action	Required topic permissions	Required group permissions
ShareConsumer read	DeveloperRead	DeveloperRead
User read queue	DeveloperRead	DeveloperRead
Delete share group	None	DeveloperManage
Delete offsets	DeveloperRead	DeveloperManage
Reset offsets	DeveloperRead	DeveloperRead
View default settings	None	DeveloperManage
Alter default settings	None	ResourceOwner

Confluent CLI

You can use the Confluent CLI to grant permissions for share groups and topics.

Grant share group permissions

To grant READ permissions to a share group, use the confluent iam rbac role-binding create command:

confluent iam rbac role-binding create \
  --principal User:u-a03bcd \
  --role DeveloperRead \
  --kafka-cluster 0000000000000000000000 \
  --resource Group:my-share-group

The command returns the role binding details:

+----------------+----------------------------+
| Principal      | User:u-a03bcd              |
| Role           | DeveloperRead              |
| KafkaCluster   | 0000000000000000000000     |
| ResourceType   | Group                      |
| Name           | my-share-group             |
| PatternType    | LITERAL                    |
+----------------+----------------------------+

Grant topic permissions

To grant READ permissions to a topic, use the confluent iam rbac role-binding create command:

confluent iam rbac role-binding create \
  --principal User:u-a03bcd \
  --role DeveloperRead \
  --kafka-cluster 0000000000000000000000 \
  --resource Topic:my-topic

The command returns the role binding details:

+----------------+----------------------------+
| Principal      | User:u-a03bcd              |
| Role           | DeveloperRead              |
| KafkaCluster   | 0000000000000000000000     |
| ResourceType   | Topic                      |
| Name           | my-topic                   |
| PatternType    | LITERAL                    |
+----------------+----------------------------+

For details, see confluent iam rbac role-binding create.

Using the KafkaShareConsumer for Share Groups

Use the org.apache.kafka.clients.consumer.KafkaShareConsumer class to consume records from a Kafka cluster using share groups. Setting up a KafkaShareConsumer involves configuration and a subscription process similar to a regular KafkaConsumer, but with key differences in how messages are handled and acknowledged.

Configuration

The KafkaShareConsumer can be instantiated using various constructors that accept either a Map<String, Object> or Properties object for configuration, optionally including key and value deserializers.

Example:

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "my-share-group"); // All consumers in the same share group will have the same group.id
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props);

Note

After creating a KafkaShareConsumer, you must call close() to prevent resource leaks.

Subscribing to Topics

Share consumers in a share group dynamically subscribe to topics using the subscribe(Collection<String> topics) method. Topic subscriptions are not incremental; a new list replaces any current assignment. Kafka delivers each message in the subscribed topics to one share consumer within the share group.

Example:

consumer.subscribe(Arrays.asList("my-topic"));

Polling for Records and Liveness

After subscribing, the share consumer joins the share group automatically when poll(Duration) is invoked. This method fetches data for the subscribed topics. To ensure consumer liveness and continued record reception, poll() must be called periodically. The share consumer sends periodic heartbeats to the broker, and if heartbeats are not sent within the share group’s session time-out, then the share consumer will be considered dead and its partitions will be reassigned.

Example:

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    // Process records
}

Record Delivery and Acknowledgement

When a share consumer in a share group fetches records using poll(), it receives available records from any matching topic-partitions. These records are acquired with a time-limited acquisition lock, making them unavailable to other share consumers during this period. The default lock duration is 30 seconds. The lock duration is controlled by the share.record.lock.duration.ms group configuration parameter. The lock is automatically released if its duration elapses, making the record available for another delivery attempt.

Share consumers can handle acquired records in these ways:

Acknowledge successful processing

Marks the record as successfully processed.

Release the record

Makes the record available for another delivery attempt.

Reject the record

Indicates the record is unprocessable and prevents further delivery attempts.

Renew the record

Requires the lock on the record for long-running processing.

Do nothing

The lock automatically releases when its duration expires.

The share consumer can choose to use implicit or explicit acknowledgement of the records it processes by using the share consumer share.acknowledgement.mode configuration property.

Implicit Acknowledgement

If the share consumer configuration property share.acknowledgement.mode is set to implicit or is not set at all, then the share consumer uses implicit acknowledgement. All delivered records are implicitly marked as successfully processed and acknowledged when commitSync(), commitAsync(), or poll() is called.

Example (Implicit with poll):

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); // Implicitly acknowledges previous batch
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        // doProcessing(record);
    }
}

Example (implicit with commitSync):

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        // doProcessing(record);
    }
    consumer.commitSync(); // Implicitly acknowledges all delivered records
}

For more information, see KafkaShareConsumer Javadocs.

Explicit Acknowledgement

If the share consumer configuration property share.acknowledgement.mode is set to explicit, then the share consumer uses explicit acknowledgement. The application must acknowledge all records returned from poll(Duration) using acknowledge(ConsumerRecord, AcknowledgeType) before its next call to poll(Duration). The AcknowledgeType argument indicates whether the record was processed successfully (ACCEPT), should be released for another attempt (RELEASE), permanently rejected (REJECT), or needs extended processing time (RENEW). Acknowledgements are committed when commitSync(), commitAsync(), or poll() is called.

The available acknowledgement types are:

ACCEPT

Marks the record as successfully processed.

RELEASE

Makes the record available for another delivery attempt.

REJECT

Indicates the record is unprocessable and prevents further delivery attempts.

RENEW

Extends the acquisition lock timeout for records that require additional processing time beyond the default lock duration. Prevents the record from being re-delivered while still being processed. Use acquisitionLockTimeoutMs() to determine when to renew the lock.

Example (Explicit acknowledgement with commitSync):

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        try {
            // doProcessing(record);
            consumer.acknowledge(record, AcknowledgeType.ACCEPT);
        } catch (Exception e) {
            consumer.acknowledge(record, AcknowledgeType.REJECT);
        }
    }
    consumer.commitSync();
}

Example (Explicit acknowledgement with RENEW for long-running processing):

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        try {
            // Check if processing exceeds acquisition lock timeout
            Optional<Integer> lockTimeout = consumer.acquisitionLockTimeoutMs();
            if (needsMoreTime(record) && lockTimeout.isPresent()) {
                // Extend the lock to prevent re-delivery during processing
                consumer.acknowledge(record, AcknowledgeType.RENEW);
                consumer.commitSync();
            }
            // doLongRunningProcessing(record);
            consumer.acknowledge(record, AcknowledgeType.ACCEPT);
        } catch (Exception e) {
            consumer.acknowledge(record, AcknowledgeType.REJECT);
        }
    }
    consumer.commitSync();
}

For more information, see KafkaShareConsumer Javadocs.

Multithreaded Processing

The KafkaShareConsumer is not thread-safe, and unsynchronized access will lead to ConcurrentModificationException. The only exception is wakeup(), which can safely be called from an external thread to interrupt an active operation (for example, poll()), causing a WakeupException to be thrown in the blocking thread. This mechanism is useful for safely shutting down the share consumer from another thread.

Example of safe shutdown using wakeup():

public class KafkaShareConsumerRunner implements Runnable {
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final KafkaShareConsumer consumer;

    public KafkaShareConsumerRunner(KafkaShareConsumer consumer) {
        this.consumer = consumer;
    }

    @Override
    public void run() {
        try {
            consumer.subscribe(Arrays.asList("topic"));
            while (!closed.get()) {
                ConsumerRecords records = consumer.poll(Duration.ofMillis(10000));
                // Handle new records
            }
        } catch (WakeupException e) {
            // Ignore exception if closing
            if (!closed.get()) throw e;
        } finally {
            consumer.close();
        }
    }

    public void shutdown() {
        closed.set(true);
        consumer.wakeup();
    }
}

Note

While thread interrupts are possible, wakeup() is the discouraged method for aborting blocking operations for a clean consumer shutdown.

Transactional Records and Isolation Level

The handling of transactional records in share groups is governed by the share.isolation.level configuration property, which applies to the entire share group.

read_uncommitted (Default)

The share group consumes all non-transactional and transactional records, bounded by the high-water mark.

read_committed

The share group consumes only non-transactional records and committed transactional records. This level is bounded by the last stable offset, meaning an open transaction can block the share group’s progress.

Related content

• Share Groups for Confluent Platform

• Configure Share Groups for Docker in Confluent Platform

• Monitor Kafka with JMX in Confluent Platform

• Monitor and Track Metrics in Docker with JMX in Confluent Platform