Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,63 @@ The `@KafkaListener` annotation now supports an `ackMode` attribute, allowing in
The attribute also supports SpEL expressions and property placeholders.
See xref:kafka/receiving-messages/listener-annotation.adoc[`@KafkaListener` Annotation] for more information.

[[x41-share-ack-mode]]
=== Share Consumer Acknowledgment Modes

The boolean `setExplicitShareAcknowledgment(boolean)` property on `ContainerProperties` has been replaced by the `ShareAckMode` enum, which clearly names the three distinct acknowledgment modes:

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That is too much info for whats-new.adoc.
Why change pattern and provide all of that here, but not in the target chapter?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, will move to the chapter.


* `EXPLICIT` (default) — Container-managed.
The container sends `ACCEPT` after successful processing and delegates error handling to the `ShareConsumerRecordRecoverer` (default: `REJECT`).
This is the analogue of disabling `auto.commit` on a regular consumer.
* `MANUAL` — Listener-managed.
The listener must acknowledge each record via the provided `ShareAcknowledgment`.
Subsequent polls are blocked until all records from the previous poll are acknowledged.
* `IMPLICIT` — Kafka client implicit mode.
The broker auto-accepts all records regardless of processing outcome.

Configure the mode on the factory:

[source,java]
----
factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);
----

The deprecated `setExplicitShareAcknowledgment(true)` maps to `MANUAL`; `setExplicitShareAcknowledgment(false)` maps to `EXPLICIT`.
See xref:kafka/kafka-queues.adoc#share-record-acknowledgment[Record Acknowledgment] for the full reference.

==== Migration Guide

**Default behavior is unchanged.**
The old `setExplicitShareAcknowledgment(false)` default was already container-managed acknowledgment (the container sent `ACCEPT` on success), which is exactly what `ShareAckMode.EXPLICIT` does.
No action is required for applications using the default.

**If you used `setExplicitShareAcknowledgment(true)`**, replace it:

[source,java]
----
// Before (4.0)
factory.getContainerProperties().setExplicitShareAcknowledgment(true);

// After (4.1)
factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);
----

**If you set `share.acknowledgement.mode=implicit` in the factory configuration** (via `ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG`), this is a breaking change.
In 4.0 this setting had no effect because the container always called `consumer.acknowledge()` regardless, which would have thrown `IllegalStateException` in true Kafka implicit mode.
In 4.1, the container detects this conflict and logs a warning, then overrides the factory setting with explicit mode.
To genuinely use Kafka client implicit mode — where the broker auto-accepts all records regardless of processing outcome — you must now opt in explicitly:

[source,java]
----
factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.IMPLICIT);
----

[WARNING]
====
In `ShareAckMode.IMPLICIT`, the `ShareConsumerRecordRecoverer` is not consulted and processing errors do not influence acknowledgment.
Records are always ACCEPTed by the broker.
====

[[x41-share-consumer-error-handling]]
=== Share Consumer Error Handling

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,6 @@
import java.util.Collection;
import java.util.regex.Pattern;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.internals.ShareAcknowledgementMode;
import org.jspecify.annotations.Nullable;

import org.springframework.beans.BeanUtils;
Expand Down Expand Up @@ -183,11 +181,6 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
BeanUtils.copyProperties(this.containerProperties, properties, "topics", "topicPartitions", "topicPattern",
"messageListener", "ackCount", "ackTime", "subBatchPerPartition", "kafkaConsumerProperties");

// Determine acknowledgment mode following Spring Kafka's configuration precedence patterns
// Check factory-level properties first, then consumer factory config
boolean explicitAck = determineExplicitAcknowledgment(properties);
properties.setExplicitShareAcknowledgment(explicitAck);

// Set concurrency - endpoint setting takes precedence over factory setting
Integer conc = endpoint.getConcurrency();
if (conc != null) {
Expand All @@ -208,39 +201,6 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
.acceptIfNotNull(endpoint.getClientIdPrefix(), properties::setClientId);
}

/**
* Determine whether explicit acknowledgment is required following Spring Kafka's configuration precedence patterns.
* <p>
* Configuration precedence (highest to lowest):
* <ol>
* <li>Container Properties: {@code containerProperties.isExplicitShareAcknowledgment()} (if explicitly set via factory-level properties)</li>
* <li>Consumer Config: {@code ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG}</li>
* <li>Default: {@code false} (implicit acknowledgment)</li>
* </ol>
* @param containerProperties the container properties to check
* @return true if explicit acknowledgment is required, false for implicit
* @throws IllegalArgumentException if an invalid acknowledgment mode is configured
*/
private boolean determineExplicitAcknowledgment(ContainerProperties containerProperties) {
// Check factory-level properties first
// If explicitly set to true (non-default), use it with highest precedence
if (this.containerProperties.isExplicitShareAcknowledgment()) {
return true;
}

// Check Kafka client configuration as fallback
Object clientAckMode = this.shareConsumerFactory.getConfigurationProperties()
.get(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG);

if (clientAckMode != null) {
ShareAcknowledgementMode mode = ShareAcknowledgementMode.fromString(clientAckMode.toString());
return mode == ShareAcknowledgementMode.EXPLICIT;
}

// Default to implicit acknowledgment (false)
return false;
}

private static void validateShareConfiguration(KafkaListenerEndpoint endpoint) {
// Validate that batch listeners aren't used with share consumers
if (Boolean.TRUE.equals(endpoint.getBatchListener())) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaShareConsumer;
import org.apache.kafka.clients.consumer.ShareConsumer;
import org.apache.kafka.common.MetricName;
Expand Down Expand Up @@ -141,19 +142,42 @@ public ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullab
return createRawConsumer(groupId, clientId);
}

@Override
public ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullable String clientId,
Map<String, Object> overrideProperties) {
return createRawConsumer(groupId, clientId, overrideProperties);
}

/**
* Actually create the consumer.
* Create the consumer with no override properties.
* Subclasses may override to customize consumer creation.
* @param groupId the group id (maybe null).
* @param clientId the client id.
* @param clientId the client id (maybe null).
* @return the share consumer.
*/
protected ShareConsumer<K, V> createRawConsumer(@Nullable String groupId, @Nullable String clientId) {
return createRawConsumer(groupId, clientId, Collections.emptyMap());
}

/**
* Actually create the consumer, applying override properties on top of the
* factory configuration. Override properties take precedence over factory
* configuration; {@code groupId} and {@code clientId} are applied last.
* @param groupId the group id (maybe null).
* @param clientId the client id (maybe null).
* @param overrideProperties properties to apply on top of the factory configuration.
* @return the share consumer.
* @since 4.1
*/
protected ShareConsumer<K, V> createRawConsumer(@Nullable String groupId, @Nullable String clientId,
Map<String, Object> overrideProperties) {
Map<String, Object> consumerProperties = new HashMap<>(this.configs);
consumerProperties.putAll(overrideProperties);
if (groupId != null) {
consumerProperties.put("group.id", groupId);
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
}
if (clientId != null) {
consumerProperties.put("client.id", clientId);
consumerProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
}
return new ExtendedShareConsumer(consumerProperties);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,25 @@ public interface ShareConsumerFactory<K, V> {
*/
ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullable String clientId);

/**
* Create a share consumer with the provided group id, client id, and
* additional properties that override the factory's configuration.
* The container uses this to enforce internal configuration (e.g. acknowledgement
* mode) without mutating the factory's configuration.
* Implementations that do not override this method will fall back to
* {@link #createShareConsumer(String, String)}, and the override properties
* will be ignored.
* @param groupId the group id (maybe null).
* @param clientId the client id (maybe null).
* @param overrideProperties properties to apply on top of the factory configuration.
* @return the share consumer.
* @since 4.1
*/
default ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullable String clientId,
Map<String, Object> overrideProperties) {
return createShareConsumer(groupId, clientId);
Comment thread
artembilan marked this conversation as resolved.
}

/**
* Return an unmodifiable reference to the configuration map for this factory.
* Useful for cloning to make a similar factory.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,16 +27,20 @@
* <p>
* This interface provides access to both the {@link ShareConsumer} instance and acknowledgment
* capabilities. The acknowledgment parameter behavior depends on the container's
* acknowledgment mode:
* {@link ContainerProperties.ShareAckMode}:
* <ul>
* <li><strong>Explicit mode</strong>: The acknowledgment parameter is non-null and must
* be used to acknowledge each record</li>
* <li><strong>Implicit mode</strong>: The acknowledgment parameter is null and records
* are automatically acknowledged</li>
* <li><strong>MANUAL</strong>: The acknowledgment is non-null; the listener must call
* {@link org.springframework.kafka.support.ShareAcknowledgment#acknowledge()},
* {@link org.springframework.kafka.support.ShareAcknowledgment#release()}, or
* {@link org.springframework.kafka.support.ShareAcknowledgment#reject()} for every record</li>
* <li><strong>EXPLICIT</strong>: The acknowledgment is null; the container sends ACCEPT
* automatically on success and delegates errors to the {@link ShareConsumerRecordRecoverer}</li>
* <li><strong>IMPLICIT</strong>: The acknowledgment is null; the broker auto-accepts all records</li>
* </ul>
* <p>
* This is the primary listener interface for share consumers when you need access
* to the ShareConsumer instance or need explicit acknowledgment control.
* to the {@link ShareConsumer} instance or need listener-managed acknowledgment
* control ({@code ShareAckMode.MANUAL}).
*
* @param <K> the key type
* @param <V> the value type
Expand All @@ -52,12 +56,11 @@
public interface AcknowledgingShareConsumerAwareMessageListener<K, V> extends GenericMessageListener<ConsumerRecord<K, V>> {

/**
* Invoked with data from kafka, an acknowledgment, and provides access to the consumer.
* When explicit acknowledgment mode is used, the acknowledgment parameter will be non-null
* and must be used to acknowledge the record. When implicit acknowledgment mode is used,
* the acknowledgment parameter will be null.
* Invoked with data from Kafka, an acknowledgment, and provides access to the consumer.
* The acknowledgment is non-null only in {@link ContainerProperties.ShareAckMode#MANUAL} mode;
* it is null in {@code EXPLICIT} and {@code IMPLICIT} modes.
* @param data the data to be processed.
* @param acknowledgment the acknowledgment (nullable in implicit mode).
* @param acknowledgment the acknowledgment, or {@code null} if not in MANUAL mode.
* @param consumer the consumer.
*/
void onShareRecord(ConsumerRecord<K, V> data, @Nullable ShareAcknowledgment acknowledgment, ShareConsumer<?, ?> consumer);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.kafka.support.ShareAcknowledgment;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.kafka.support.micrometer.KafkaListenerObservationConvention;
import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
Expand Down Expand Up @@ -117,6 +116,40 @@ public enum AckMode {

}

/**
* The acknowledgment mode for share consumer containers.
* @since 4.1
* @see #setShareAckMode(ShareAckMode)
*/
public enum ShareAckMode {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Feels like a decision to reuse ContainerProperties for ShareConsumer was wrong direction.
Now we give end-user confusion with existing AckMode, which apparently is not used ShareKafkaMessageListenerContainer.
And there are probably many other conflicting options not used here or there.
Not saying that it should be revised now, but something to keep in mind if we'd like to still keep our API end-user friendly.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ya, we have time to refine it before RC1.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The problem is that we need to distinguish two kinds of containers if we introduce a properties class just for share consumers.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Artem - as we discussed yesterday, we will try to tackle this as a follow up item as this involves a lot of moving parts.


/**
* Kafka client implicit mode. All records are automatically acknowledged
* as ACCEPT by the broker regardless of processing outcome. No per-record
* acknowledgment control is available in this mode. Equivalent to setting
* {@code share.acknowledgement.mode=implicit} on the Kafka client.
*/
IMPLICIT,

/**
* Kafka client explicit mode, container-managed. The container sends ACCEPT
* after successful processing and delegates to the
* {@link ShareConsumerRecordRecoverer} (default: REJECT) on error.
* This is the default.
*/
EXPLICIT,

/**
* Kafka client explicit mode, listener-managed. The listener must acknowledge
* each record manually via the provided
* {@link org.springframework.kafka.support.ShareAcknowledgment}.
* Subsequent polls are blocked until all records from the previous poll
* are acknowledged.
*/
MANUAL

}

/**
* Offset commit behavior during assignment.
* @since 2.3.6
Expand Down Expand Up @@ -314,7 +347,7 @@ public enum EOSMode {

private boolean recordObservationsInBatch;

private boolean explicitShareAcknowledgment = false;
private ShareAckMode shareAckMode = ShareAckMode.EXPLICIT; // default: container-managed explicit mode

private Duration shareAcknowledgmentTimeout = Duration.ofSeconds(30); // Align with Kafka's share.record.lock.duration.ms default

Expand Down Expand Up @@ -1121,38 +1154,54 @@ public void setRecordObservationsInBatch(boolean recordObservationsInBatch) {
}

/**
* Set whether explicit acknowledgment is required for share consumer containers.
* Set the acknowledgment mode for share consumer containers.
* <p>
* This setting only applies to share consumer containers and is ignored
* by regular consumer containers.
* <p>
* When set to {@code false} (default), records are automatically acknowledged
* as ACCEPT when the next poll occurs or when commitSync/commitAsync is called.
* <p>
* When set to {@code true}, the application must explicitly acknowledge each
* record using the provided {@link ShareAcknowledgment}.
* @param explicitShareAcknowledgment true for explicit acknowledgment, false for implicit
* @since 4.0
* @see ShareAcknowledgment
* @param shareAckMode the acknowledgment mode; default {@link ShareAckMode#EXPLICIT}.
* @since 4.1
* @see ShareAckMode
*/
public void setShareAckMode(ShareAckMode shareAckMode) {
this.shareAckMode = shareAckMode;
}

/**
* Return the acknowledgment mode for share consumer containers.
* @return the acknowledgment mode.
* @since 4.1
*/
public ShareAckMode getShareAckMode() {
return this.shareAckMode;
}

/**
* Set whether to use explicit share acknowledgment mode.
* @param explicitShareAcknowledgment {@code true} to use {@link ShareAckMode#MANUAL},
* {@code false} to use {@link ShareAckMode#EXPLICIT}.
* @deprecated in favor of {@link #setShareAckMode(ShareAckMode)} with
* {@link ShareAckMode#MANUAL}.
*/
@Deprecated(since = "4.1", forRemoval = false)
public void setExplicitShareAcknowledgment(boolean explicitShareAcknowledgment) {
this.explicitShareAcknowledgment = explicitShareAcknowledgment;
this.shareAckMode = explicitShareAcknowledgment ? ShareAckMode.MANUAL : ShareAckMode.EXPLICIT;
}

/**
* Check whether explicit acknowledgment is required for share consumer containers.
* @return true if explicit acknowledgment is required, false for implicit acknowledgment
* Return whether the current mode is {@link ShareAckMode#MANUAL}.
* @return {@code true} if the current {@link ShareAckMode} is {@link ShareAckMode#MANUAL}.
* @deprecated in favor of {@link #getShareAckMode()}.
*/
@Deprecated(since = "4.1", forRemoval = false)
public boolean isExplicitShareAcknowledgment() {
return this.explicitShareAcknowledgment;
return this.shareAckMode == ShareAckMode.MANUAL;
}

/**
* Set the timeout for share acknowledgments in explicit mode.
* Set the timeout for share acknowledgments in {@link ShareAckMode#MANUAL} mode.
* <p>
* When a record is not acknowledged within this timeout, a warning
* will be logged to help identify missing acknowledgment calls.
* This only applies when using explicit acknowledgment mode.
* When a record is not acknowledged within this timeout, a warning will be logged
* to help identify missing acknowledgment calls. Only applies to {@code MANUAL} mode.
* <p>
* Default is 30 seconds.
* @param shareAcknowledgmentTimeout the timeout duration
Expand All @@ -1163,7 +1212,7 @@ public void setShareAcknowledgmentTimeout(Duration shareAcknowledgmentTimeout) {
}

/**
* Get the timeout for share acknowledgments in explicit mode.
* Get the timeout for share acknowledgments in {@link ShareAckMode#MANUAL} mode.
* @return the acknowledgment timeout
* @since 4.0
*/
Expand Down
Loading