You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "TaiJuWu (via GitHub)" <gi...@apache.org> on 2026/02/17 02:51:38 UTC
[PR] KAFKA-20159: Group configuration for share.delivery.count.limit [kafka]
TaiJuWu opened a new pull request, #21491:
URL: https://github.com/apache/kafka/pull/21491
Make `share.delivery.count.limit` become dynamic config, we can change the value at run time and set to specific group.
--
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-20159: Group configuration for share.delivery.count.limit [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #21491:
URL: https://github.com/apache/kafka/pull/21491#discussion_r2824203304
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1951,7 +1947,7 @@ private int acquireSubsetBatchRecords(
// records in a single response batch. Condition below checks if the current record has reached
// the delivery limit and already have some records to return in response then skip processing
// the current record, which shall be delivered alone in next fetch.
- if (maxDeliveryCount > 2 && recordDeliveryCount == maxDeliveryCount - 1 && acquiredCount > 0) {
+ if (maxDeliveryCount() > 2 && recordDeliveryCount == maxDeliveryCount() - 1 && acquiredCount > 0) {
Review Comment:
I'd prefer you to obtain the values of `maxDeliveryCount()` and `throttleRecordsDeliveryLimit()` at the start of the method and then use those values throughout. There is a tiny chance that the configs change during the execution of the method and I think there's a minuscule chance of a peculiar bug emerging.
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -359,6 +359,7 @@ class KafkaApisTest extends Logging {
cgConfigs.put(SHARE_SESSION_TIMEOUT_MS_CONFIG, GroupCoordinatorConfig.SHARE_GROUP_SESSION_TIMEOUT_MS_DEFAULT.toString)
cgConfigs.put(SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, GroupCoordinatorConfig.SHARE_GROUP_HEARTBEAT_INTERVAL_MS_DEFAULT.toString)
cgConfigs.put(SHARE_RECORD_LOCK_DURATION_MS_CONFIG, ShareGroupConfig.SHARE_GROUP_RECORD_LOCK_DURATION_MS_DEFAULT.toString)
+ cgConfigs.put(SHARE_DELIVERY_COUNT_LIMIT_CONFIG, ShareGroupConfig.SHARE_GROUP_MAX_DELIVERY_COUNT_LIMIT_DEFAULT.toString)
Review Comment:
I think `ShareGroupConfig.SHARE_GROUP_MAX_DELIVERY_COUNT_LIMIT_DEFAULT` should be `ShareGroupConfig.SHARE_GROUP_DELIVERY_COUNT_LIMIT_DEFAULT` here.
--
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-20159: Group configuration for share.delivery.count.limit [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #21491:
URL: https://github.com/apache/kafka/pull/21491#discussion_r2825135410
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1951,7 +1947,7 @@ private int acquireSubsetBatchRecords(
// records in a single response batch. Condition below checks if the current record has reached
// the delivery limit and already have some records to return in response then skip processing
// the current record, which shall be delivered alone in next fetch.
- if (maxDeliveryCount > 2 && recordDeliveryCount == maxDeliveryCount - 1 && acquiredCount > 0) {
+ if (maxDeliveryCount() > 2 && recordDeliveryCount == maxDeliveryCount() - 1 && acquiredCount > 0) {
Review Comment:
Got it, thanks for your review.
--
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-20159: Group configuration for share.delivery.count.limit [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #21491:
URL: https://github.com/apache/kafka/pull/21491#discussion_r2825135847
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -359,6 +359,7 @@ class KafkaApisTest extends Logging {
cgConfigs.put(SHARE_SESSION_TIMEOUT_MS_CONFIG, GroupCoordinatorConfig.SHARE_GROUP_SESSION_TIMEOUT_MS_DEFAULT.toString)
cgConfigs.put(SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, GroupCoordinatorConfig.SHARE_GROUP_HEARTBEAT_INTERVAL_MS_DEFAULT.toString)
cgConfigs.put(SHARE_RECORD_LOCK_DURATION_MS_CONFIG, ShareGroupConfig.SHARE_GROUP_RECORD_LOCK_DURATION_MS_DEFAULT.toString)
+ cgConfigs.put(SHARE_DELIVERY_COUNT_LIMIT_CONFIG, ShareGroupConfig.SHARE_GROUP_MAX_DELIVERY_COUNT_LIMIT_DEFAULT.toString)
Review Comment:
fixed it.
--
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-20159: Group configuration for share.delivery.count.limit [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #21491:
URL: https://github.com/apache/kafka/pull/21491#discussion_r2825135410
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1951,7 +1947,7 @@ private int acquireSubsetBatchRecords(
// records in a single response batch. Condition below checks if the current record has reached
// the delivery limit and already have some records to return in response then skip processing
// the current record, which shall be delivered alone in next fetch.
- if (maxDeliveryCount > 2 && recordDeliveryCount == maxDeliveryCount - 1 && acquiredCount > 0) {
+ if (maxDeliveryCount() > 2 && recordDeliveryCount == maxDeliveryCount() - 1 && acquiredCount > 0) {
Review Comment:
Got it, thanks for your point out.
--
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-20159: Group configuration for share.delivery.count.limit [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #21491:
URL: https://github.com/apache/kafka/pull/21491#discussion_r2817334276
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupConfig.java:
##########
Review Comment:
I like the "value of a group configuration for" language. Could you please apply it for `SHARE_GROUP_MIN_RECORD_LOCK_DURATION_MS_DOC` and `SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS_DOC`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java:
##########
@@ -259,6 +271,14 @@ private static void validateValues(Map<?, ?> valueMaps, GroupCoordinatorConfig g
throw new InvalidConfigurationException(SHARE_RECORD_LOCK_DURATION_MS_CONFIG + " must be less than or equal to " +
ShareGroupConfig.SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS_CONFIG);
}
+ if (shareDeliveryCountLimit < shareGroupConfig.shareGroupMinDeliveryCountLimit()) {
+ throw new InvalidConfigurationException(SHARE_DELIVERY_COUNT_LIMIT_CONFIG + " must be greater than or equal to " +
+ ShareGroupConfig.SHARE_GROUP_MIN_DELIVERY_COUNT_LIMIT_CONFIG);
+ }
+ if (shareDeliveryCountLimit > shareGroupConfig.shareGroupMaxDeliveryCountLimit()) {
+ throw new InvalidConfigurationException(SHARE_DELIVERY_COUNT_LIMIT_CONFIG + " must be greater than or equal to " +
Review Comment:
must be LESS than
--
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-20159: Group configuration for share.delivery.count.limit [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #21491:
URL: https://github.com/apache/kafka/pull/21491#discussion_r2820650750
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java:
##########
@@ -259,6 +271,14 @@ private static void validateValues(Map<?, ?> valueMaps, GroupCoordinatorConfig g
throw new InvalidConfigurationException(SHARE_RECORD_LOCK_DURATION_MS_CONFIG + " must be less than or equal to " +
ShareGroupConfig.SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS_CONFIG);
}
+ if (shareDeliveryCountLimit < shareGroupConfig.shareGroupMinDeliveryCountLimit()) {
+ throw new InvalidConfigurationException(SHARE_DELIVERY_COUNT_LIMIT_CONFIG + " must be greater than or equal to " +
+ ShareGroupConfig.SHARE_GROUP_MIN_DELIVERY_COUNT_LIMIT_CONFIG);
+ }
+ if (shareDeliveryCountLimit > shareGroupConfig.shareGroupMaxDeliveryCountLimit()) {
+ throw new InvalidConfigurationException(SHARE_DELIVERY_COUNT_LIMIT_CONFIG + " must be greater than or equal to " +
Review Comment:
Fixed it.
--
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-20159: Group configuration for share.delivery.count.limit [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #21491:
URL: https://github.com/apache/kafka/pull/21491
--
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