You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "DL1231 (via GitHub)" <gi...@apache.org> on 2026/02/13 03:52:22 UTC
[PR] KAFKA-20037: Group configuration for share.renew.acknowledge.enable [kafka]
DL1231 opened a new pull request, #21467:
URL: https://github.com/apache/kafka/pull/21467
Introduce the `share.renew.acknowledge.enable` group configuration (KIP-1222) to control whether renew acknowledgements are permitted for a share group.
- Add new boolean dynamic group config `share.renew.acknowledge.enable` (default false) in `GroupConfig`
- When disabled, RENEW acknowledgements are rejected with `INVALID_RECORD_STATE` error code
- Update existing RENEW tests to explicitly enable the config
- Add unit tests and integration test for the disabled-by-default behavior
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2889941592
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2298,6 +2303,11 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
byte ackType = ackTypeMap.size() > 1 ? ackTypeMap.get(offsetState.getKey()) : batch.acknowledgeTypes().get(0);
if (ackType == AcknowledgeType.RENEW.id) {
+ if (!configProvider.isRenewAcknowledgeEnabled(groupId)) {
+ log.warn("Renew acknowledge is not enabled for the group: {}", groupId);
Review Comment:
I'm not sure if `WARN` is appropriate here, as other cases that return `InvalidRecordStateException` use the `DEBUG` level. Should we stay consistent with those?
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -3082,6 +3082,36 @@ public void testRenewAcknowledgementInvalidStateRecord() {
verifyYammerMetricCount("ackType=Renew", 0);
}
+ @ClusterTest
+ public void testRenewAcknowledgementDisabled() {
+ alterShareAutoOffsetReset("group1", "earliest");
+ alterShareRenewAcknowledgeEnable("group1", false);
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ "group1",
+ Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT))
+ ) {
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), "Message".getBytes());
+ producer.send(record);
+ producer.flush();
+
+ shareConsumer.subscribe(List.of(tp.topic()));
+ ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer, 2500L, 1);
+ assertEquals(1, records.count());
+
+ for (ConsumerRecord<byte[], byte[]> rec : records) {
+ shareConsumer.acknowledge(rec, AcknowledgeType.RENEW);
+ }
+
+ // The RENEW should be rejected by the broker because share.renew.acknowledge.enable defaults to false.
Review Comment:
The default value is `true`, right?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2371,6 +2381,11 @@ private Optional<Throwable> acknowledgeCompleteBatch(
// Before reaching this point, it should be verified that it is full batch ack and
// not per offset ack as well as startOffset not moved.
if (ackType == AcknowledgeType.RENEW.id) {
+ if (!configProvider.isRenewAcknowledgeEnabled(groupId)) {
+ log.warn("Renew acknowledge is not enabled for the group: {}", groupId);
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #21467:
URL: https://github.com/apache/kafka/pull/21467
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2802643908
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -275,6 +276,20 @@ public static int recordLockDurationMsOrDefault(GroupConfigManager groupConfigMa
return defaultValue;
}
+ /**
+ * The method is used to check if renew acknowledge is enabled for the group. If the group config
+ * is present, then the value from the group config is used. Otherwise, the default value is used.
+ *
+ * @param groupConfigManager The group config manager.
+ * @param groupId The group id for which the renew acknowledge enable is to be checked.
+ * @return true if renew acknowledge is enabled for the group, false otherwise.
+ */
+ public static boolean isRenewAcknowledgeEnabled(GroupConfigManager groupConfigManager, String groupId) {
Review Comment:
Moving this to `GroupMetadataManager` would require introducing `GroupMetadataManager` as a new dependency of `SharePartition`, which seems like unnecessary coupling. The similar config-accessor methods in
`GroupMetadataManager` (e.g., `consumerGroupSessionTimeoutMs`) serve the coordinator's own internal logic, whereas this helper serves the share partition layer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2844038557
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -275,6 +276,20 @@ public static int recordLockDurationMsOrDefault(GroupConfigManager groupConfigMa
return defaultValue;
}
+ /**
+ * The method is used to check if renew acknowledge is enabled for the group. If the group config
+ * is present, then the value from the group config is used. Otherwise, the default value is used.
+ *
+ * @param groupConfigManager The group config manager.
+ * @param groupId The group id for which the renew acknowledge enable is to be checked.
+ * @return true if renew acknowledge is enabled for the group, false otherwise.
+ */
+ public static boolean isRenewAcknowledgeEnabled(GroupConfigManager groupConfigManager, String groupId) {
Review Comment:
`ShareFetchUtils` has other helpers following the same pattern (`recordLockDurationMsOrDefault`, `deliveryCountLimitOrDefault`), moving just this one would be inconsistent, and moving all of them would add
share-specific surface to `GroupConfigManager`, which currently has a clean CRUD-style API in the group-coordinator module.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2874848912
##########
core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java:
##########
@@ -747,6 +748,24 @@ void testDeliveryCountLimitOrDefaultWithoutGroupConfig() {
assertEquals(5, ShareFetchUtils.deliveryCountLimitOrDefault(groupConfigManager, "test-group", 5));
}
+ @Test
+ void testIsNewAcknowledgeDisabledWithGroupConfig() {
Review Comment:
This should be `testIsRenewAcknowledgeDisabledWithGroupConfig`. There's no such thing as "NewAcknowledge". :)
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -275,6 +276,20 @@ public static int recordLockDurationMsOrDefault(GroupConfigManager groupConfigMa
return defaultValue;
}
+ /**
+ * The method is used to check if renew acknowledge is enabled for the group. If the group config
+ * is present, then the value from the group config is used. Otherwise, the default value is used.
+ *
+ * @param groupConfigManager The group config manager.
+ * @param groupId The group id for which the renew acknowledge enable is to be checked.
+ * @return true if renew acknowledge is enabled for the group, false otherwise.
+ */
+ public static boolean isRenewAcknowledgeEnabled(GroupConfigManager groupConfigManager, String groupId) {
Review Comment:
@chia7712 Are you happy with this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2844050439
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java:
##########
@@ -69,6 +70,10 @@ public final class GroupConfig extends AbstractConfig {
"If set to \"read_uncommitted\", the share group will return all messages, even transactional messages which have been aborted. " +
"Non-transactional records will be returned unconditionally in either mode.";
+ public static final String SHARE_RENEW_ACKNOWLEDGE_ENABLE_CONFIG = "share.renew.acknowledge.enable";
+ public static final boolean SHARE_RENEW_ACKNOWLEDGE_ENABLE_DEFAULT = false;
Review Comment:
Thanks for catching that, updated to `true`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2890088737
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2298,6 +2303,11 @@ private Optional<Throwable> acknowledgePerOffsetBatchRecords(
byte ackType = ackTypeMap.size() > 1 ? ackTypeMap.get(offsetState.getKey()) : batch.acknowledgeTypes().get(0);
if (ackType == AcknowledgeType.RENEW.id) {
+ if (!configProvider.isRenewAcknowledgeEnabled(groupId)) {
+ log.warn("Renew acknowledge is not enabled for the group: {}", groupId);
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #21467:
URL: https://github.com/apache/kafka/pull/21467#issuecomment-3998117861
@DL1231 There are a couple of tests failing now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2808248082
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -275,6 +276,20 @@ public static int recordLockDurationMsOrDefault(GroupConfigManager groupConfigMa
return defaultValue;
}
+ /**
+ * The method is used to check if renew acknowledge is enabled for the group. If the group config
+ * is present, then the value from the group config is used. Otherwise, the default value is used.
+ *
+ * @param groupConfigManager The group config manager.
+ * @param groupId The group id for which the renew acknowledge enable is to be checked.
+ * @return true if renew acknowledge is enabled for the group, false otherwise.
+ */
+ public static boolean isRenewAcknowledgeEnabled(GroupConfigManager groupConfigManager, String groupId) {
Review Comment:
How about `GroupConfigManager`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2802340842
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -275,6 +276,20 @@ public static int recordLockDurationMsOrDefault(GroupConfigManager groupConfigMa
return defaultValue;
}
+ /**
+ * The method is used to check if renew acknowledge is enabled for the group. If the group config
+ * is present, then the value from the group config is used. Otherwise, the default value is used.
+ *
+ * @param groupConfigManager The group config manager.
+ * @param groupId The group id for which the renew acknowledge enable is to be checked.
+ * @return true if renew acknowledge is enabled for the group, false otherwise.
+ */
+ public static boolean isRenewAcknowledgeEnabled(GroupConfigManager groupConfigManager, String groupId) {
Review Comment:
Should we move this to `GroupMetadataManager`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2814282044
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java:
##########
@@ -69,6 +70,10 @@ public final class GroupConfig extends AbstractConfig {
"If set to \"read_uncommitted\", the share group will return all messages, even transactional messages which have been aborted. " +
"Non-transactional records will be returned unconditionally in either mode.";
+ public static final String SHARE_RENEW_ACKNOWLEDGE_ENABLE_CONFIG = "share.renew.acknowledge.enable";
+ public static final boolean SHARE_RENEW_ACKNOWLEDGE_ENABLE_DEFAULT = false;
Review Comment:
The default in KIP-1240 is `true`. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20161 Group configuration for share.renew.acknowledge.enable [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21467:
URL: https://github.com/apache/kafka/pull/21467#discussion_r2876851661
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -275,6 +276,20 @@ public static int recordLockDurationMsOrDefault(GroupConfigManager groupConfigMa
return defaultValue;
}
+ /**
+ * The method is used to check if renew acknowledge is enabled for the group. If the group config
+ * is present, then the value from the group config is used. Otherwise, the default value is used.
+ *
+ * @param groupConfigManager The group config manager.
+ * @param groupId The group id for which the renew acknowledge enable is to be checked.
+ * @return true if renew acknowledge is enabled for the group, false otherwise.
+ */
+ public static boolean isRenewAcknowledgeEnabled(GroupConfigManager groupConfigManager, String groupId) {
Review Comment:
I'm a bit concerned that `ShareFetchUtils` is accumulating too many helpers and might eventually become a god object. Maybe we could introduces a `ShareGroupConfigProvider` class to encapsulate these config-related helpers?
```java
public class ShareGroupConfigProvider {
private final GroupConfigManager manager;
public ShareGroupConfigProvider(GroupConfigManager manager) {
this.manager = manager;
}
public boolean isRenewAcknowledgeEnabled(String groupId) {
return manager.groupConfig(groupId)
.map(GroupConfig::shareRenewAcknowledgeEnable)
.orElse(GroupConfig.SHARE_RENEW_ACKNOWLEDGE_ENABLE_DEFAULT);
}
}
```
Also, would you mind refactoring `deliveryCountLimitOrDefault` and `recordLockDurationMsOrDefault`? They don't use the fluent style, which could potentially produce a `NoSuchElementException` if the config is removed concurrently from the `GroupConfigManager`
```java
public static int recordLockDurationMsOrDefault(GroupConfigManager groupConfigManager, String groupId, int defaultValue) {
if (groupConfigManager.groupConfig(groupId).isPresent()) {
// if the config gets removed here, it causes NoSuchElementException
return groupConfigManager.groupConfig(groupId).get().shareRecordLockDurationMs();
}
return defaultValue;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org