You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dajac (via GitHub)" <gi...@apache.org> on 2023/03/01 13:35:21 UTC
[GitHub] [kafka] dajac opened a new pull request, #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
dajac opened a new pull request, #13322:
URL: https://github.com/apache/kafka/pull/13322
This patch adds the new server configurations introduced in KIP-848. All of them are kept as internal configurations for now. We will make them public when the KIP is ready. It also adds an internal configuration named `group.coordinator.new.enable` that we will used during the development of the KIP. This one will be remove later on and replaced by IBP/feature flag.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[GitHub] [kafka] dajac commented on pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #13322:
URL: https://github.com/apache/kafka/pull/13322#issuecomment-1450613278
> Just a general question about configs -- do we have any guidelines for adding and removing configs between releases? Just curious about compatibility and typical guidelines around that.
Configs are treated as public APIs so we treat them as such. This is why I keep all of them internal for now. They won't be published until we remove 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
[GitHub] [kafka] dajac commented on pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #13322:
URL: https://github.com/apache/kafka/pull/13322#issuecomment-1451584302
@jolshan @jeffkbkim Thanks for your comments. Addressed them.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13322:
URL: https://github.com/apache/kafka/pull/13322#discussion_r1123513945
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1267,6 +1314,24 @@ object KafkaConfig {
.define(GroupInitialRebalanceDelayMsProp, INT, Defaults.GroupInitialRebalanceDelayMs, MEDIUM, GroupInitialRebalanceDelayMsDoc)
.define(GroupMaxSizeProp, INT, Defaults.GroupMaxSize, atLeast(1), MEDIUM, GroupMaxSizeDoc)
+ /** New group coordinator configs */
+ // All properties are kept internal until KIP-848 is released.
+ // This property meant to be here only during the development of KIP-848. It will
+ // be replaced by a metadata version before releasing it.
+ .defineInternal(NewGroupCoordinatorEnableProp, BOOLEAN, Defaults.NewGroupCoordinatorEnable, null, MEDIUM, NewGroupCoordinatorEnableDoc)
Review Comment:
ah interesting. Thanks for clarifying.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13322:
URL: https://github.com/apache/kafka/pull/13322#discussion_r1122808060
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1267,6 +1314,24 @@ object KafkaConfig {
.define(GroupInitialRebalanceDelayMsProp, INT, Defaults.GroupInitialRebalanceDelayMs, MEDIUM, GroupInitialRebalanceDelayMsDoc)
.define(GroupMaxSizeProp, INT, Defaults.GroupMaxSize, atLeast(1), MEDIUM, GroupMaxSizeDoc)
+ /** New group coordinator configs */
+ // All properties are kept internal until KIP-848 is released.
+ // This property meant to be here only during the development of KIP-848. It will
+ // be replaced by a metadata version before releasing it.
+ .defineInternal(NewGroupCoordinatorEnableProp, BOOLEAN, Defaults.NewGroupCoordinatorEnable, null, MEDIUM, NewGroupCoordinatorEnableDoc)
+ .defineInternal(GroupCoordinatorNumThreadsProp, INT, Defaults.GroupCoordinatorNumThreads, atLeast(1), MEDIUM, GroupCoordinatorNumThreadsDoc)
+
+ /** Consumer groups configs */
+ // All properties are kept internal until KIP-848 is released.
+ .defineInternal(ConsumerGroupSessionTimeoutMsProp, INT, Defaults.ConsumerGroupSessionTimeoutMs, atLeast(1), MEDIUM, ConsumerGroupSessionTimeoutMsDoc)
Review Comment:
They will be mainly used to ensure that the dynamic group configurations are within the defined boundaries.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13322:
URL: https://github.com/apache/kafka/pull/13322#discussion_r1122120725
##########
core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala:
##########
@@ -976,6 +976,20 @@ class KafkaConfigTest {
case RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
case RemoteLogManagerConfig.REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
+ /** New group coordinator configs */
+ case KafkaConfig.NewGroupCoordinatorEnableProp => // ignore string
+ case KafkaConfig.GroupCoordinatorNumThreadsProp => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
+
+ /** Consumer groups configs */
+ case KafkaConfig.ConsumerGroupSessionTimeoutMsProp => // ignore string
Review Comment:
Should this one assert the same as the ones below.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13322:
URL: https://github.com/apache/kafka/pull/13322#discussion_r1122119145
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1267,6 +1314,24 @@ object KafkaConfig {
.define(GroupInitialRebalanceDelayMsProp, INT, Defaults.GroupInitialRebalanceDelayMs, MEDIUM, GroupInitialRebalanceDelayMsDoc)
.define(GroupMaxSizeProp, INT, Defaults.GroupMaxSize, atLeast(1), MEDIUM, GroupMaxSizeDoc)
+ /** New group coordinator configs */
+ // All properties are kept internal until KIP-848 is released.
+ // This property meant to be here only during the development of KIP-848. It will
+ // be replaced by a metadata version before releasing it.
+ .defineInternal(NewGroupCoordinatorEnableProp, BOOLEAN, Defaults.NewGroupCoordinatorEnable, null, MEDIUM, NewGroupCoordinatorEnableDoc)
+ .defineInternal(GroupCoordinatorNumThreadsProp, INT, Defaults.GroupCoordinatorNumThreads, atLeast(1), MEDIUM, GroupCoordinatorNumThreadsDoc)
+
+ /** Consumer groups configs */
+ // All properties are kept internal until KIP-848 is released.
+ .defineInternal(ConsumerGroupSessionTimeoutMsProp, INT, Defaults.ConsumerGroupSessionTimeoutMs, atLeast(1), MEDIUM, ConsumerGroupSessionTimeoutMsDoc)
+ .defineInternal(ConsumerGroupMinSessionTimeoutMsProp, INT, Defaults.ConsumerGroupMinSessionTimeoutMs, atLeast(1), MEDIUM, ConsumerGroupMinSessionTimeoutMsDoc)
+ .defineInternal(ConsumerGroupMaxSessionTimeoutMsProp, INT, Defaults.ConsumerGroupMaxSessionTimeoutMs, atLeast(1), MEDIUM, ConsumerGroupMaxSessionTimeoutMsDoc)
Review Comment:
Do we want to enforce that the min value is less than or equal to the max value?
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13322:
URL: https://github.com/apache/kafka/pull/13322#discussion_r1122805573
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1267,6 +1314,24 @@ object KafkaConfig {
.define(GroupInitialRebalanceDelayMsProp, INT, Defaults.GroupInitialRebalanceDelayMs, MEDIUM, GroupInitialRebalanceDelayMsDoc)
.define(GroupMaxSizeProp, INT, Defaults.GroupMaxSize, atLeast(1), MEDIUM, GroupMaxSizeDoc)
+ /** New group coordinator configs */
+ // All properties are kept internal until KIP-848 is released.
+ // This property meant to be here only during the development of KIP-848. It will
+ // be replaced by a metadata version before releasing it.
+ .defineInternal(NewGroupCoordinatorEnableProp, BOOLEAN, Defaults.NewGroupCoordinatorEnable, null, MEDIUM, NewGroupCoordinatorEnableDoc)
Review Comment:
`defineInternal` does not have an overload without the `validator`.
--
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
[GitHub] [kafka] dajac merged pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #13322:
URL: https://github.com/apache/kafka/pull/13322
--
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
[GitHub] [kafka] jolshan commented on pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #13322:
URL: https://github.com/apache/kafka/pull/13322#issuecomment-1450610551
Just a general question about configs -- do we have any guidelines for adding and removing configs between releases? Just curious about compatibility and typical guidelines around that.
--
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
[GitHub] [kafka] jolshan commented on pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #13322:
URL: https://github.com/apache/kafka/pull/13322#issuecomment-1450638812
Got it. So the distinction is "publishing" and that is done when the config is not internal. Thanks for clarifying.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13322:
URL: https://github.com/apache/kafka/pull/13322#discussion_r1122800222
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -163,6 +163,20 @@ object Defaults {
val GroupInitialRebalanceDelayMs = 3000
val GroupMaxSize: Int = Int.MaxValue
+ /** New group coordinator configs */
+ val NewGroupCoordinatorEnable = false
+ val GroupCoordinatorNumThreads = 1
+
+ /** Consumer group configs */
+ val ConsumerGroupSessionTimeoutMs = 45000
+ val ConsumerGroupMinSessionTimeoutMs = 45000
+ val ConsumerGroupMaxSessionTimeoutMs = 60000
+ val ConsumerGroupHeartbeatIntervalMs = 5000
+ val ConsumerGroupMinHeartbeatInternalMs = 5000
+ val ConsumerGroupMaxHeartbeatInternalMs = 15000
+ val ConsumerGroupMaxSize = Int.MaxValue
+ val ConsumerGroupAssignors = ""
Review Comment:
Correct. We will add them when we have them.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13322:
URL: https://github.com/apache/kafka/pull/13322#discussion_r1122123411
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1267,6 +1314,24 @@ object KafkaConfig {
.define(GroupInitialRebalanceDelayMsProp, INT, Defaults.GroupInitialRebalanceDelayMs, MEDIUM, GroupInitialRebalanceDelayMsDoc)
.define(GroupMaxSizeProp, INT, Defaults.GroupMaxSize, atLeast(1), MEDIUM, GroupMaxSizeDoc)
+ /** New group coordinator configs */
+ // All properties are kept internal until KIP-848 is released.
+ // This property meant to be here only during the development of KIP-848. It will
+ // be replaced by a metadata version before releasing it.
+ .defineInternal(NewGroupCoordinatorEnableProp, BOOLEAN, Defaults.NewGroupCoordinatorEnable, null, MEDIUM, NewGroupCoordinatorEnableDoc)
Review Comment:
nit: you can leave "null" out (same with the other prop below)
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13322:
URL: https://github.com/apache/kafka/pull/13322#discussion_r1122121405
##########
core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala:
##########
@@ -976,6 +976,20 @@ class KafkaConfigTest {
case RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
case RemoteLogManagerConfig.REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
+ /** New group coordinator configs */
+ case KafkaConfig.NewGroupCoordinatorEnableProp => // ignore string
Review Comment:
nit: this one isn't a string -- some of the other configs just said "ignore"
--
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
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #13322: KAFKA-14462; [1/N] Add new server configurations (KIP-848)
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #13322:
URL: https://github.com/apache/kafka/pull/13322#discussion_r1122374811
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -163,6 +163,20 @@ object Defaults {
val GroupInitialRebalanceDelayMs = 3000
val GroupMaxSize: Int = Int.MaxValue
+ /** New group coordinator configs */
+ val NewGroupCoordinatorEnable = false
+ val GroupCoordinatorNumThreads = 1
+
+ /** Consumer group configs */
+ val ConsumerGroupSessionTimeoutMs = 45000
+ val ConsumerGroupMinSessionTimeoutMs = 45000
+ val ConsumerGroupMaxSessionTimeoutMs = 60000
+ val ConsumerGroupHeartbeatIntervalMs = 5000
+ val ConsumerGroupMinHeartbeatInternalMs = 5000
+ val ConsumerGroupMaxHeartbeatInternalMs = 15000
+ val ConsumerGroupMaxSize = Int.MaxValue
+ val ConsumerGroupAssignors = ""
Review Comment:
KIP 848 lists
```
org.apache.kafka.server.group.consumer.UniformAssignor, org.apache.kafka.server.group.consumer.RangeAssignor
```
as the default. are we waiting until these two are implemented?
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -163,6 +163,20 @@ object Defaults {
val GroupInitialRebalanceDelayMs = 3000
val GroupMaxSize: Int = Int.MaxValue
+ /** New group coordinator configs */
+ val NewGroupCoordinatorEnable = false
+ val GroupCoordinatorNumThreads = 1
+
+ /** Consumer group configs */
+ val ConsumerGroupSessionTimeoutMs = 45000
+ val ConsumerGroupMinSessionTimeoutMs = 45000
+ val ConsumerGroupMaxSessionTimeoutMs = 60000
+ val ConsumerGroupHeartbeatIntervalMs = 5000
+ val ConsumerGroupMinHeartbeatInternalMs = 5000
+ val ConsumerGroupMaxHeartbeatInternalMs = 15000
Review Comment:
should these be "Interval"?
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -483,11 +497,27 @@ object KafkaConfig {
val ControlledShutdownMaxRetriesProp = "controlled.shutdown.max.retries"
val ControlledShutdownRetryBackoffMsProp = "controlled.shutdown.retry.backoff.ms"
val ControlledShutdownEnableProp = "controlled.shutdown.enable"
+
/** ********* Group coordinator configuration ***********/
val GroupMinSessionTimeoutMsProp = "group.min.session.timeout.ms"
val GroupMaxSessionTimeoutMsProp = "group.max.session.timeout.ms"
val GroupInitialRebalanceDelayMsProp = "group.initial.rebalance.delay.ms"
val GroupMaxSizeProp = "group.max.size"
+
+ /** New group coordinator configs */
+ val NewGroupCoordinatorEnableProp = "group.coordinator.new.enable"
+ val GroupCoordinatorNumThreadsProp = "group.coordinator.threads"
+
+ /** Consumer group configs */
+ val ConsumerGroupSessionTimeoutMsProp = "group.consumer.session.timeout.ms"
+ val ConsumerGroupMinSessionTimeoutMsProp = "group.consumer.min.session.timeout.ms"
+ val ConsumerGroupMaxSessionTimeoutMsProp = "group.consumer.max.session.timeout.ms"
+ val ConsumerGroupHeartbeatIntervalMsProp = "group.consumer.heartbeat.interval.ms"
+ val ConsumerGroupMinHeartbeatInternalMsProp = "group.consumer.min.heartbeat.interval.ms"
+ val ConsumerGroupMaxHeartbeatInternalMsProp ="group.consumer.max.heartbeat.interval.ms"
Review Comment:
should these also be "Interval"?
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1267,6 +1314,24 @@ object KafkaConfig {
.define(GroupInitialRebalanceDelayMsProp, INT, Defaults.GroupInitialRebalanceDelayMs, MEDIUM, GroupInitialRebalanceDelayMsDoc)
.define(GroupMaxSizeProp, INT, Defaults.GroupMaxSize, atLeast(1), MEDIUM, GroupMaxSizeDoc)
+ /** New group coordinator configs */
+ // All properties are kept internal until KIP-848 is released.
+ // This property meant to be here only during the development of KIP-848. It will
+ // be replaced by a metadata version before releasing it.
+ .defineInternal(NewGroupCoordinatorEnableProp, BOOLEAN, Defaults.NewGroupCoordinatorEnable, null, MEDIUM, NewGroupCoordinatorEnableDoc)
+ .defineInternal(GroupCoordinatorNumThreadsProp, INT, Defaults.GroupCoordinatorNumThreads, atLeast(1), MEDIUM, GroupCoordinatorNumThreadsDoc)
+
+ /** Consumer groups configs */
+ // All properties are kept internal until KIP-848 is released.
+ .defineInternal(ConsumerGroupSessionTimeoutMsProp, INT, Defaults.ConsumerGroupSessionTimeoutMs, atLeast(1), MEDIUM, ConsumerGroupSessionTimeoutMsDoc)
Review Comment:
how does the operator use ConsumerGroupSessionTimeoutMs and ConsumerGroupSessionMin/MaxTimeoutMs configs?
is the idea to set min/max once, then configure the base timeout which the min/max would enforce?
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1267,6 +1314,24 @@ object KafkaConfig {
.define(GroupInitialRebalanceDelayMsProp, INT, Defaults.GroupInitialRebalanceDelayMs, MEDIUM, GroupInitialRebalanceDelayMsDoc)
.define(GroupMaxSizeProp, INT, Defaults.GroupMaxSize, atLeast(1), MEDIUM, GroupMaxSizeDoc)
+ /** New group coordinator configs */
+ // All properties are kept internal until KIP-848 is released.
+ // This property meant to be here only during the development of KIP-848. It will
Review Comment:
nit: "property is meant"
--
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