You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "clolov (via GitHub)" <gi...@apache.org> on 2026/04/02 13:35:13 UTC
[PR] KAFKA-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
clolov opened a new pull request, #21943:
URL: https://github.com/apache/kafka/pull/21943
This PR moves the bulk of remaining KafkaConfig configs from Scala to Java.
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3949922686
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -245,101 +203,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
}
}
- /***************** rack configuration **************/
- val replicaSelectorClassName = Option(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG))
-
- /** ********* Replication configuration ***********/
- val replicaLagTimeMaxMs = getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG)
Review Comment:
Good point! I tried moving all of these in the constructor, but one can't use getInt, getLong etc. before KafkaConfig.currentConfig is initialised. I think I can see a way to do it, but that would be a separate PR on which this one will build upon. The other approach I can think of is to wrap everything in code similar to the quotaConfig i.e. creating it on a first usage and from then on returning the reference. I traced the usages of a few of the configs and they appear to be used during startup or/and only once and do not appear to be called on every request. As such, if we leave them as this PR suggests right now there may be a regression, but it doesn't seem to affect traffic.
What would you feel more comfortable with proceeding, the current version, a wrapper similar to what we do today with quotaConfig or trying to refactor KafkaConfig.currentConfig in AbstractKafkaConfig in order to be able to use getInt, getLong etc. in the constructor? Or maybe you can see another way?
##########
core/src/main/scala/kafka/server/ControllerServer.scala:
##########
@@ -235,7 +235,7 @@ class ControllerServer(
OptionalLong.empty()
}
- val maxIdleIntervalNs = config.metadataMaxIdleIntervalNs.fold(OptionalLong.empty)(OptionalLong.of)
+ val maxIdleIntervalNs = config.metadataMaxIdleIntervalNs.map[OptionalLong](v => OptionalLong.of(v)).orElse(OptionalLong.empty)
Review Comment:
Great spot!
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3031223349
##########
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java:
##########
@@ -45,7 +45,7 @@
*/
public class AbstractConfig {
- private static final Logger log = LoggerFactory.getLogger(AbstractConfig.class);
+ public static final Logger log = LoggerFactory.getLogger(AbstractConfig.class);
Review Comment:
consider protected instead of public, or a dedicated logger in AbstractKafkaConfig, to avoid expanding AbstractConfig’s public surface.
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -635,4 +642,295 @@ public Map<String, Object> extractGroupConfigMap() {
);
return defaults;
}
+
+ // ********* General Configuration **********
+
+ public int brokerSessionTimeoutMs() {
+ return getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG);
+ }
+
+ public long controllerPerformanceSamplePeriodMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS);
+ }
+
+ public long controllerPerformanceAlwaysLogThresholdMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS);
+ }
+
+ public Set<ProcessRole> parseProcessRoles() {
+ List<String> roles = getList(KRaftConfigs.PROCESS_ROLES_CONFIG);
+ Set<ProcessRole> result = new HashSet<>();
+ for (String role : roles) {
+ switch (role) {
+ case "broker" -> result.add(ProcessRole.BrokerRole);
+ case "controller" -> result.add(ProcessRole.ControllerRole);
+ default -> throw new ConfigException("Unknown process role '" + role +
+ "' (only 'broker' and 'controller' are allowed roles)");
+ }
+ }
+ return Collections.unmodifiableSet(result);
+ }
+
+ public boolean isKRaftCombinedMode(Set<ProcessRole> processRoles) {
+ return processRoles.equals(Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole));
+ }
+
+ public String metadataLogDir() {
+ String dir = getString(MetadataLogConfig.METADATA_LOG_DIR_CONFIG);
+ return dir != null ? dir : logDirs().get(0);
+ }
+
+ public long serverMaxStartupTimeMs() {
+ return getLong(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG);
+ }
+
+ public Integer messageMaxBytes() {
+ return getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
+ }
+
+ public int getNumReplicaAlterLogDirsThreads() {
+ Integer numThreads = getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG);
+ return numThreads != null ? numThreads : logDirs().size();
+ }
+
+ // ********* Metadata Configuration **********
+
+ public long metadataSnapshotMaxNewRecordBytes() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG);
+ }
+
+ public long metadataSnapshotMaxIntervalMs() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG);
+ }
+
+ public Optional<Long> metadataMaxIdleIntervalNs() {
+ long value = TimeUnit.NANOSECONDS.convert(
+ getInt(MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG).longValue(),
+ TimeUnit.MILLISECONDS);
+ return value > 0 ? Optional.of(value) : Optional.empty();
+ }
+
+ // ********* Rack Configuration **********
+
+ public Optional<String> replicaSelectorClassName() {
+ return Optional.ofNullable(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG));
+ }
+
+ // ********* Replication Configuration **********
+
+ public int controllerSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.CONTROLLER_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int defaultReplicationFactor() {
+ return getInt(ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG);
+ }
+
+ public long replicaLagTimeMaxMs() {
+ return getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG);
+ }
+
+ public int replicaSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int replicaSocketReceiveBufferBytes() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_RECEIVE_BUFFER_BYTES_CONFIG);
+ }
+
+ public int replicaFetchMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchWaitMaxMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG);
+ }
+
+ public int replicaFetchMinBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MIN_BYTES_CONFIG);
+ }
+
+ public int replicaFetchResponseMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchBackoffMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_BACKOFF_MS_CONFIG);
+ }
+
+ public long replicaHighWatermarkCheckpointIntervalMs() {
+ return getLong(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG);
+ }
+
+ public int fetchPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int producerPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.PRODUCER_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int deleteRecordsPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.DELETE_RECORDS_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public boolean autoLeaderRebalanceEnable() {
+ return getBoolean(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG);
+ }
+
+ public long leaderImbalanceCheckIntervalSeconds() {
+ return getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG);
+ }
+
+ public long uncleanLeaderElectionCheckIntervalMs() {
+ return getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG);
+ }
+
+ public Boolean uncleanLeaderElectionEnable() {
+ return getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG);
+ }
+
+ public Boolean followerFetchLastTieredOffsetEnable() {
+ return getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
+ }
+
+ // ********* Controlled Shutdown Configuration **********
+
+ public boolean controlledShutdownEnable() {
+ return getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG);
+ }
+
+ // ********* Group Coordinator Configuration **********
+
+ @SuppressWarnings("removal")
+ public Set<GroupType> groupCoordinatorRebalanceProtocols() {
Review Comment:
Moving groupCoordinatorRebalanceProtocols from a Scala val to a Java method re-runs validation/log.warn on every access—consider caching to match one-time init 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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3950088498
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -245,101 +203,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
}
}
- /***************** rack configuration **************/
- val replicaSelectorClassName = Option(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG))
-
- /** ********* Replication configuration ***********/
- val replicaLagTimeMaxMs = getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG)
Review Comment:
Fair enough. Would you mind opening a ticket for it? The performance regression should be tiny, so we can defer the optimization. Let's finish the migration first.
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3952880118
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -174,48 +170,10 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
private val _shareCoordinatorConfig = new ShareCoordinatorConfig(this)
def shareCoordinatorConfig: ShareCoordinatorConfig = _shareCoordinatorConfig
- /** ********* General Configuration ***********/
- val brokerSessionTimeoutMs: Int = getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG)
- val controllerPerformanceSamplePeriodMs: Long = getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS)
- val controllerPerformanceAlwaysLogThresholdMs: Long = getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS)
val controllerMaxRecordsPerBatch: Int = getInt(KRaftConfigs.CONTROLLER_MAX_RECORDS_PER_BATCH_CONFIG)
- private def parseProcessRoles(): Set[ProcessRole] = {
- val roles = getList(KRaftConfigs.PROCESS_ROLES_CONFIG).asScala.map {
- case "broker" => ProcessRole.BrokerRole
- case "controller" => ProcessRole.ControllerRole
- case role => throw new ConfigException(s"Unknown process role '$role'" +
- " (only 'broker' and 'controller' are allowed roles)")
- }
- roles.toSet
- }
-
def isKRaftCombinedMode: Boolean = {
Review Comment:
This method is used by testing only, so maybe we could just remove 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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3968337990
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -679,4 +691,294 @@ public Map<String, Object> extractGroupConfigMap(GroupCoordinatorConfig groupCoo
);
return defaults;
}
+
+ // ********* General Configuration **********
+
+ public int brokerSessionTimeoutMs() {
+ return getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG);
+ }
+
+ public long controllerPerformanceSamplePeriodMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS);
+ }
+
+ public long controllerPerformanceAlwaysLogThresholdMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS);
+ }
+
+ public Set<ProcessRole> processRoles() {
+ Set<ProcessRole> value = processRoles;
+ if (value == null) {
+ Set<ProcessRole> result = new HashSet<>();
+ for (String role : getList(KRaftConfigs.PROCESS_ROLES_CONFIG)) {
+ switch (role) {
+ case "broker" -> result.add(ProcessRole.BrokerRole);
+ case "controller" -> result.add(ProcessRole.ControllerRole);
+ default -> throw new ConfigException("Unknown process role '" + role +
+ "' (only 'broker' and 'controller' are allowed roles)");
+ }
+ }
+ value = Collections.unmodifiableSet(result);
+ processRoles = value;
+ }
+ return value;
+ }
+
+ public boolean isKRaftCombinedMode(Set<ProcessRole> processRoles) {
+ return processRoles.equals(Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole));
+ }
+
+ public String metadataLogDir() {
+ String dir = getString(MetadataLogConfig.METADATA_LOG_DIR_CONFIG);
+ return dir != null ? dir : logDirs().get(0);
+ }
+
+ public long serverMaxStartupTimeMs() {
+ return getLong(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG);
+ }
+
+ public Integer messageMaxBytes() {
+ return getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
+ }
+
+ public int getNumReplicaAlterLogDirsThreads() {
+ Integer numThreads = getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG);
+ return numThreads != null ? numThreads : logDirs().size();
+ }
+
+ // ********* Metadata Configuration **********
+
+ public long metadataSnapshotMaxNewRecordBytes() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG);
+ }
+
+ public long metadataSnapshotMaxIntervalMs() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG);
+ }
+
+ public OptionalLong metadataMaxIdleIntervalNs() {
+ long value = TimeUnit.NANOSECONDS.convert(
+ getInt(MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG).longValue(),
+ TimeUnit.MILLISECONDS);
+ return value > 0 ? OptionalLong.of(value) : OptionalLong.empty();
+ }
+
+ // ********* Rack Configuration **********
+
+ public Optional<String> replicaSelectorClassName() {
+ return Optional.ofNullable(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG));
+ }
+
+ // ********* Replication Configuration **********
+
+ public long replicaLagTimeMaxMs() {
+ return getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG);
+ }
+
+ public int replicaSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int replicaSocketReceiveBufferBytes() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_RECEIVE_BUFFER_BYTES_CONFIG);
+ }
+
+ public int replicaFetchMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchWaitMaxMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG);
+ }
+
+ public int replicaFetchMinBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MIN_BYTES_CONFIG);
+ }
+
+ public int replicaFetchResponseMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchBackoffMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_BACKOFF_MS_CONFIG);
+ }
+
+ public long replicaHighWatermarkCheckpointIntervalMs() {
+ return getLong(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG);
+ }
+
+ public int fetchPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int producerPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.PRODUCER_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int deleteRecordsPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.DELETE_RECORDS_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public boolean autoLeaderRebalanceEnable() {
+ return getBoolean(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG);
+ }
+
+ public long leaderImbalanceCheckIntervalSeconds() {
+ return getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG);
+ }
+
+ public long uncleanLeaderElectionCheckIntervalMs() {
+ return getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG);
+ }
+
+ public Boolean uncleanLeaderElectionEnable() {
+ return getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG);
+ }
+
+ public Boolean followerFetchLastTieredOffsetEnable() {
+ return getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
+ }
+
+ // ********* Controlled Shutdown Configuration **********
+
+ public boolean controlledShutdownEnable() {
+ return getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG);
+ }
+
+ // ********* Group Coordinator Configuration **********
+
+ @SuppressWarnings("removal")
+ public Set<GroupType> groupCoordinatorRebalanceProtocols() {
+ return getList(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)
+ .stream()
+ .map(s -> GroupType.valueOf(s.toUpperCase(Locale.ROOT)))
+ .collect(Collectors.toUnmodifiableSet());
+ }
+
+ @SuppressWarnings("removal")
+ protected void validateGroupCoordinatorRebalanceProtocols() {
+ Set<GroupType> protocols = groupCoordinatorRebalanceProtocols();
+
+ if (!protocols.contains(GroupType.CLASSIC)) {
+ throw new ConfigException("Disabling the '" + GroupType.CLASSIC + "' protocol is not supported.");
+ }
+
+ if (doLog && protocols.contains(GroupType.SHARE)) {
+ log.warn("'{}' in `{}` is deprecated. " +
+ "Share groups are controlled by the 'share.version' feature. " +
+ "This config will be removed in Kafka 5.0.",
+ GroupType.SHARE,
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG);
+ }
+
+ if (doLog && originals().containsKey(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)) {
+ Set<GroupType> defaultProtocols = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT
+ .stream()
+ .map(s -> GroupType.valueOf(s.toUpperCase(Locale.ROOT)))
+ .collect(Collectors.toUnmodifiableSet());
+
+ Set<GroupType> missingProtocols = new HashSet<>(defaultProtocols);
+ missingProtocols.removeAll(protocols);
+
+ if (!missingProtocols.isEmpty()) {
+ log.warn("The config `{}` is deprecated and will be removed in Kafka 5.0. " +
+ "The following protocol(s) are currently disabled: {}. " +
+ "In Kafka 5.0, all protocols will always be enabled and controlled solely by feature versions " +
+ "(group.version, streams.version, share.version) via kafka-features.sh. " +
+ "Please remove the configuration, which will restore all protocols to the default enabled state, to prepare for the upgrade.",
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG,
+ missingProtocols.stream().map(GroupType::toString).collect(Collectors.joining(", ")));
+ } else {
+ log.warn("The config `{}` is deprecated and will be removed in Kafka 5.0. " +
+ "Please remove the configuration to prepare for the upgrade.",
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG);
+ }
+ }
+ }
+
+ // ********* Metric Configuration **********
+
+ public int metricNumSamples() {
+ return getInt(MetricConfigs.METRIC_NUM_SAMPLES_CONFIG);
+ }
+
+ public long metricSampleWindowMs() {
+ return getLong(MetricConfigs.METRIC_SAMPLE_WINDOW_MS_CONFIG);
+ }
+
+ public String metricRecordingLevel() {
+ return getString(MetricConfigs.METRIC_RECORDING_LEVEL_CONFIG);
+ }
+
+ // ********* Kafka Client Telemetry Metrics Configuration **********
+
+ public int clientTelemetryMaxBytes() {
+ return getInt(MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_CONFIG);
+ }
+
+ // ********* SSL/SASL Configuration **********
+ // Security configs may be overridden for listeners, so it is not safe to use the base values.
+ // Hence the base SSL/SASL configs are not fields of KafkaConfig, listener configs should be
+ // retrieved using KafkaConfig#valuesWithPrefixOverride
+
+ @SuppressWarnings("unchecked")
+ public Set<String> saslEnabledMechanisms(ListenerName listenerName) {
Review Comment:
Great catch!
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3977945904
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -90,9 +99,12 @@ public abstract class AbstractKafkaConfig extends AbstractConfig {
));
private volatile QuotaConfig quotaConfig;
+ private volatile Set<ProcessRole> processRoles;
Review Comment:
As part of this, I learned something new today. Spotbugs ought to catch things like this, but it doesn't in abstract classes. I find this a bit interesting, because abstract or not-abstract shouldn't matter for a private field, but I guess it does. Thanks for spotting this! The new revision corrects 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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3952874584
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -2248,7 +2248,7 @@ class ReplicaManager(val config: KafkaConfig,
}
private def createReplicaSelector(metrics: Metrics): Option[Plugin[ReplicaSelector]] = {
- config.replicaSelectorClassName.map { className =>
+ config.replicaSelectorClassName.asScala.map { className =>
Review Comment:
`toScala`
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3937232978
##########
core/src/main/scala/kafka/server/ControllerServer.scala:
##########
@@ -235,7 +235,7 @@ class ControllerServer(
OptionalLong.empty()
}
- val maxIdleIntervalNs = config.metadataMaxIdleIntervalNs.fold(OptionalLong.empty)(OptionalLong.of)
+ val maxIdleIntervalNs = config.metadataMaxIdleIntervalNs.map[OptionalLong](v => OptionalLong.of(v)).orElse(OptionalLong.empty)
Review Comment:
Maybe `metadataMaxIdleIntervalNs` could return `OptionalLong` instead?
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -245,101 +203,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
}
}
- /***************** rack configuration **************/
- val replicaSelectorClassName = Option(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG))
-
- /** ********* Replication configuration ***********/
- val replicaLagTimeMaxMs = getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG)
Review Comment:
I'm wondering if those val fields should stay as fields in the Java version. Will evaluating them on each call be a performance regression?
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3957943865
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -125,7 +121,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
// Cache the current config to avoid acquiring read lock to access from dynamicConfig
@volatile private var currentConfig = this
- val processRoles: Set[ProcessRole] = parseProcessRoles()
+ val processRoles: Set[ProcessRole] = parseProcessRoles().asScala.toSet
Review Comment:
I hopefully addressed it the way you were thinking about 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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3962099133
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -552,7 +411,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
validateControllerQuorumVotersMustContainNodeIdForKRaftController()
validateAdvertisedControllerListenersNonEmptyForKRaftController()
validateControllerListenerNamesMustAppearInListenersForKRaftController()
- } else if (isKRaftCombinedMode) {
+ } else if (isKRaftCombinedMode(processRoles)) {
Review Comment:
```scala
processRoles == util.Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole)
```
We could inline it and remove the method, since this is its only usage :)
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -679,4 +691,294 @@ public Map<String, Object> extractGroupConfigMap(GroupCoordinatorConfig groupCoo
);
return defaults;
}
+
+ // ********* General Configuration **********
+
+ public int brokerSessionTimeoutMs() {
+ return getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG);
+ }
+
+ public long controllerPerformanceSamplePeriodMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS);
+ }
+
+ public long controllerPerformanceAlwaysLogThresholdMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS);
+ }
+
+ public Set<ProcessRole> processRoles() {
+ Set<ProcessRole> value = processRoles;
+ if (value == null) {
+ Set<ProcessRole> result = new HashSet<>();
+ for (String role : getList(KRaftConfigs.PROCESS_ROLES_CONFIG)) {
+ switch (role) {
+ case "broker" -> result.add(ProcessRole.BrokerRole);
+ case "controller" -> result.add(ProcessRole.ControllerRole);
+ default -> throw new ConfigException("Unknown process role '" + role +
+ "' (only 'broker' and 'controller' are allowed roles)");
+ }
+ }
+ value = Collections.unmodifiableSet(result);
+ processRoles = value;
+ }
+ return value;
+ }
+
+ public boolean isKRaftCombinedMode(Set<ProcessRole> processRoles) {
+ return processRoles.equals(Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole));
+ }
+
+ public String metadataLogDir() {
+ String dir = getString(MetadataLogConfig.METADATA_LOG_DIR_CONFIG);
+ return dir != null ? dir : logDirs().get(0);
+ }
+
+ public long serverMaxStartupTimeMs() {
+ return getLong(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG);
+ }
+
+ public Integer messageMaxBytes() {
+ return getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
+ }
+
+ public int getNumReplicaAlterLogDirsThreads() {
+ Integer numThreads = getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG);
+ return numThreads != null ? numThreads : logDirs().size();
+ }
+
+ // ********* Metadata Configuration **********
+
+ public long metadataSnapshotMaxNewRecordBytes() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG);
+ }
+
+ public long metadataSnapshotMaxIntervalMs() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG);
+ }
+
+ public OptionalLong metadataMaxIdleIntervalNs() {
+ long value = TimeUnit.NANOSECONDS.convert(
+ getInt(MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG).longValue(),
+ TimeUnit.MILLISECONDS);
+ return value > 0 ? OptionalLong.of(value) : OptionalLong.empty();
+ }
+
+ // ********* Rack Configuration **********
+
+ public Optional<String> replicaSelectorClassName() {
+ return Optional.ofNullable(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG));
+ }
+
+ // ********* Replication Configuration **********
+
+ public long replicaLagTimeMaxMs() {
+ return getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG);
+ }
+
+ public int replicaSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int replicaSocketReceiveBufferBytes() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_RECEIVE_BUFFER_BYTES_CONFIG);
+ }
+
+ public int replicaFetchMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchWaitMaxMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG);
+ }
+
+ public int replicaFetchMinBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MIN_BYTES_CONFIG);
+ }
+
+ public int replicaFetchResponseMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchBackoffMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_BACKOFF_MS_CONFIG);
+ }
+
+ public long replicaHighWatermarkCheckpointIntervalMs() {
+ return getLong(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG);
+ }
+
+ public int fetchPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int producerPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.PRODUCER_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int deleteRecordsPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.DELETE_RECORDS_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public boolean autoLeaderRebalanceEnable() {
+ return getBoolean(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG);
+ }
+
+ public long leaderImbalanceCheckIntervalSeconds() {
+ return getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG);
+ }
+
+ public long uncleanLeaderElectionCheckIntervalMs() {
+ return getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG);
+ }
+
+ public Boolean uncleanLeaderElectionEnable() {
+ return getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG);
+ }
+
+ public Boolean followerFetchLastTieredOffsetEnable() {
+ return getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
+ }
+
+ // ********* Controlled Shutdown Configuration **********
+
+ public boolean controlledShutdownEnable() {
+ return getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG);
+ }
+
+ // ********* Group Coordinator Configuration **********
+
+ @SuppressWarnings("removal")
+ public Set<GroupType> groupCoordinatorRebalanceProtocols() {
+ return getList(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)
+ .stream()
+ .map(s -> GroupType.valueOf(s.toUpperCase(Locale.ROOT)))
+ .collect(Collectors.toUnmodifiableSet());
+ }
+
+ @SuppressWarnings("removal")
+ protected void validateGroupCoordinatorRebalanceProtocols() {
Review Comment:
We could let `validateGroupCoordinatorRebalanceProtocols` accept a `doLog` argument, so we don't need to store `doLog` as a field.
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -90,9 +99,12 @@ public abstract class AbstractKafkaConfig extends AbstractConfig {
));
private volatile QuotaConfig quotaConfig;
+ private volatile Set<ProcessRole> processRoles;
Review Comment:
I thought we agreed to defer the caching optimization to a follow-up? It's not a big deal, but I don't see the point of caching `processRoles` here since it's not on any hot path.
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -679,4 +691,294 @@ public Map<String, Object> extractGroupConfigMap(GroupCoordinatorConfig groupCoo
);
return defaults;
}
+
+ // ********* General Configuration **********
+
+ public int brokerSessionTimeoutMs() {
+ return getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG);
+ }
+
+ public long controllerPerformanceSamplePeriodMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS);
+ }
+
+ public long controllerPerformanceAlwaysLogThresholdMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS);
+ }
+
+ public Set<ProcessRole> processRoles() {
+ Set<ProcessRole> value = processRoles;
+ if (value == null) {
+ Set<ProcessRole> result = new HashSet<>();
+ for (String role : getList(KRaftConfigs.PROCESS_ROLES_CONFIG)) {
+ switch (role) {
+ case "broker" -> result.add(ProcessRole.BrokerRole);
+ case "controller" -> result.add(ProcessRole.ControllerRole);
+ default -> throw new ConfigException("Unknown process role '" + role +
+ "' (only 'broker' and 'controller' are allowed roles)");
+ }
+ }
+ value = Collections.unmodifiableSet(result);
+ processRoles = value;
+ }
+ return value;
+ }
+
+ public boolean isKRaftCombinedMode(Set<ProcessRole> processRoles) {
+ return processRoles.equals(Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole));
+ }
+
+ public String metadataLogDir() {
+ String dir = getString(MetadataLogConfig.METADATA_LOG_DIR_CONFIG);
+ return dir != null ? dir : logDirs().get(0);
+ }
+
+ public long serverMaxStartupTimeMs() {
+ return getLong(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG);
+ }
+
+ public Integer messageMaxBytes() {
+ return getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
+ }
+
+ public int getNumReplicaAlterLogDirsThreads() {
+ Integer numThreads = getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG);
+ return numThreads != null ? numThreads : logDirs().size();
+ }
+
+ // ********* Metadata Configuration **********
+
+ public long metadataSnapshotMaxNewRecordBytes() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG);
+ }
+
+ public long metadataSnapshotMaxIntervalMs() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG);
+ }
+
+ public OptionalLong metadataMaxIdleIntervalNs() {
+ long value = TimeUnit.NANOSECONDS.convert(
+ getInt(MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG).longValue(),
+ TimeUnit.MILLISECONDS);
+ return value > 0 ? OptionalLong.of(value) : OptionalLong.empty();
+ }
+
+ // ********* Rack Configuration **********
+
+ public Optional<String> replicaSelectorClassName() {
+ return Optional.ofNullable(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG));
+ }
+
+ // ********* Replication Configuration **********
+
+ public long replicaLagTimeMaxMs() {
+ return getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG);
+ }
+
+ public int replicaSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int replicaSocketReceiveBufferBytes() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_RECEIVE_BUFFER_BYTES_CONFIG);
+ }
+
+ public int replicaFetchMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchWaitMaxMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG);
+ }
+
+ public int replicaFetchMinBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MIN_BYTES_CONFIG);
+ }
+
+ public int replicaFetchResponseMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchBackoffMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_BACKOFF_MS_CONFIG);
+ }
+
+ public long replicaHighWatermarkCheckpointIntervalMs() {
+ return getLong(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG);
+ }
+
+ public int fetchPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int producerPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.PRODUCER_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int deleteRecordsPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.DELETE_RECORDS_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public boolean autoLeaderRebalanceEnable() {
+ return getBoolean(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG);
+ }
+
+ public long leaderImbalanceCheckIntervalSeconds() {
+ return getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG);
+ }
+
+ public long uncleanLeaderElectionCheckIntervalMs() {
+ return getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG);
+ }
+
+ public Boolean uncleanLeaderElectionEnable() {
+ return getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG);
+ }
+
+ public Boolean followerFetchLastTieredOffsetEnable() {
+ return getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
+ }
+
+ // ********* Controlled Shutdown Configuration **********
+
+ public boolean controlledShutdownEnable() {
+ return getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG);
+ }
+
+ // ********* Group Coordinator Configuration **********
+
+ @SuppressWarnings("removal")
+ public Set<GroupType> groupCoordinatorRebalanceProtocols() {
+ return getList(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)
+ .stream()
+ .map(s -> GroupType.valueOf(s.toUpperCase(Locale.ROOT)))
+ .collect(Collectors.toUnmodifiableSet());
+ }
+
+ @SuppressWarnings("removal")
+ protected void validateGroupCoordinatorRebalanceProtocols() {
+ Set<GroupType> protocols = groupCoordinatorRebalanceProtocols();
+
+ if (!protocols.contains(GroupType.CLASSIC)) {
+ throw new ConfigException("Disabling the '" + GroupType.CLASSIC + "' protocol is not supported.");
+ }
+
+ if (doLog && protocols.contains(GroupType.SHARE)) {
+ log.warn("'{}' in `{}` is deprecated. " +
+ "Share groups are controlled by the 'share.version' feature. " +
+ "This config will be removed in Kafka 5.0.",
+ GroupType.SHARE,
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG);
+ }
+
+ if (doLog && originals().containsKey(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)) {
+ Set<GroupType> defaultProtocols = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT
+ .stream()
+ .map(s -> GroupType.valueOf(s.toUpperCase(Locale.ROOT)))
+ .collect(Collectors.toUnmodifiableSet());
+
+ Set<GroupType> missingProtocols = new HashSet<>(defaultProtocols);
+ missingProtocols.removeAll(protocols);
+
+ if (!missingProtocols.isEmpty()) {
+ log.warn("The config `{}` is deprecated and will be removed in Kafka 5.0. " +
+ "The following protocol(s) are currently disabled: {}. " +
+ "In Kafka 5.0, all protocols will always be enabled and controlled solely by feature versions " +
+ "(group.version, streams.version, share.version) via kafka-features.sh. " +
+ "Please remove the configuration, which will restore all protocols to the default enabled state, to prepare for the upgrade.",
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG,
+ missingProtocols.stream().map(GroupType::toString).collect(Collectors.joining(", ")));
+ } else {
+ log.warn("The config `{}` is deprecated and will be removed in Kafka 5.0. " +
+ "Please remove the configuration to prepare for the upgrade.",
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG);
+ }
+ }
+ }
+
+ // ********* Metric Configuration **********
+
+ public int metricNumSamples() {
+ return getInt(MetricConfigs.METRIC_NUM_SAMPLES_CONFIG);
+ }
+
+ public long metricSampleWindowMs() {
+ return getLong(MetricConfigs.METRIC_SAMPLE_WINDOW_MS_CONFIG);
+ }
+
+ public String metricRecordingLevel() {
+ return getString(MetricConfigs.METRIC_RECORDING_LEVEL_CONFIG);
+ }
+
+ // ********* Kafka Client Telemetry Metrics Configuration **********
+
+ public int clientTelemetryMaxBytes() {
+ return getInt(MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_CONFIG);
+ }
+
+ // ********* SSL/SASL Configuration **********
+ // Security configs may be overridden for listeners, so it is not safe to use the base values.
+ // Hence the base SSL/SASL configs are not fields of KafkaConfig, listener configs should be
+ // retrieved using KafkaConfig#valuesWithPrefixOverride
+
+ @SuppressWarnings("unchecked")
+ public Set<String> saslEnabledMechanisms(ListenerName listenerName) {
Review Comment:
`protected`
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3968357136
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -90,9 +99,12 @@ public abstract class AbstractKafkaConfig extends AbstractConfig {
));
private volatile QuotaConfig quotaConfig;
+ private volatile Set<ProcessRole> processRoles;
Review Comment:
Valid point, remove the caching!
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -90,9 +99,12 @@ public abstract class AbstractKafkaConfig extends AbstractConfig {
));
private volatile QuotaConfig quotaConfig;
+ private volatile Set<ProcessRole> processRoles;
Review Comment:
Valid point, removed the caching!
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21943:
URL: https://github.com/apache/kafka/pull/21943#issuecomment-4911409575
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3052145937
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -635,4 +642,295 @@ public Map<String, Object> extractGroupConfigMap() {
);
return defaults;
}
+
+ // ********* General Configuration **********
+
+ public int brokerSessionTimeoutMs() {
+ return getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG);
+ }
+
+ public long controllerPerformanceSamplePeriodMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS);
+ }
+
+ public long controllerPerformanceAlwaysLogThresholdMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS);
+ }
+
+ public Set<ProcessRole> parseProcessRoles() {
+ List<String> roles = getList(KRaftConfigs.PROCESS_ROLES_CONFIG);
+ Set<ProcessRole> result = new HashSet<>();
+ for (String role : roles) {
+ switch (role) {
+ case "broker" -> result.add(ProcessRole.BrokerRole);
+ case "controller" -> result.add(ProcessRole.ControllerRole);
+ default -> throw new ConfigException("Unknown process role '" + role +
+ "' (only 'broker' and 'controller' are allowed roles)");
+ }
+ }
+ return Collections.unmodifiableSet(result);
+ }
+
+ public boolean isKRaftCombinedMode(Set<ProcessRole> processRoles) {
+ return processRoles.equals(Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole));
+ }
+
+ public String metadataLogDir() {
+ String dir = getString(MetadataLogConfig.METADATA_LOG_DIR_CONFIG);
+ return dir != null ? dir : logDirs().get(0);
+ }
+
+ public long serverMaxStartupTimeMs() {
+ return getLong(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG);
+ }
+
+ public Integer messageMaxBytes() {
+ return getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
+ }
+
+ public int getNumReplicaAlterLogDirsThreads() {
+ Integer numThreads = getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG);
+ return numThreads != null ? numThreads : logDirs().size();
+ }
+
+ // ********* Metadata Configuration **********
+
+ public long metadataSnapshotMaxNewRecordBytes() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG);
+ }
+
+ public long metadataSnapshotMaxIntervalMs() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG);
+ }
+
+ public Optional<Long> metadataMaxIdleIntervalNs() {
+ long value = TimeUnit.NANOSECONDS.convert(
+ getInt(MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG).longValue(),
+ TimeUnit.MILLISECONDS);
+ return value > 0 ? Optional.of(value) : Optional.empty();
+ }
+
+ // ********* Rack Configuration **********
+
+ public Optional<String> replicaSelectorClassName() {
+ return Optional.ofNullable(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG));
+ }
+
+ // ********* Replication Configuration **********
+
+ public int controllerSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.CONTROLLER_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int defaultReplicationFactor() {
+ return getInt(ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG);
+ }
+
+ public long replicaLagTimeMaxMs() {
+ return getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG);
+ }
+
+ public int replicaSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int replicaSocketReceiveBufferBytes() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_RECEIVE_BUFFER_BYTES_CONFIG);
+ }
+
+ public int replicaFetchMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchWaitMaxMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG);
+ }
+
+ public int replicaFetchMinBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MIN_BYTES_CONFIG);
+ }
+
+ public int replicaFetchResponseMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchBackoffMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_BACKOFF_MS_CONFIG);
+ }
+
+ public long replicaHighWatermarkCheckpointIntervalMs() {
+ return getLong(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG);
+ }
+
+ public int fetchPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int producerPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.PRODUCER_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int deleteRecordsPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.DELETE_RECORDS_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public boolean autoLeaderRebalanceEnable() {
+ return getBoolean(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG);
+ }
+
+ public long leaderImbalanceCheckIntervalSeconds() {
+ return getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG);
+ }
+
+ public long uncleanLeaderElectionCheckIntervalMs() {
+ return getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG);
+ }
+
+ public Boolean uncleanLeaderElectionEnable() {
+ return getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG);
+ }
+
+ public Boolean followerFetchLastTieredOffsetEnable() {
+ return getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
+ }
+
+ // ********* Controlled Shutdown Configuration **********
+
+ public boolean controlledShutdownEnable() {
+ return getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG);
+ }
+
+ // ********* Group Coordinator Configuration **********
+
+ @SuppressWarnings("removal")
+ public Set<GroupType> groupCoordinatorRebalanceProtocols() {
Review Comment:
Also a fair suggestion!
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3968337990
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -679,4 +691,294 @@ public Map<String, Object> extractGroupConfigMap(GroupCoordinatorConfig groupCoo
);
return defaults;
}
+
+ // ********* General Configuration **********
+
+ public int brokerSessionTimeoutMs() {
+ return getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG);
+ }
+
+ public long controllerPerformanceSamplePeriodMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS);
+ }
+
+ public long controllerPerformanceAlwaysLogThresholdMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS);
+ }
+
+ public Set<ProcessRole> processRoles() {
+ Set<ProcessRole> value = processRoles;
+ if (value == null) {
+ Set<ProcessRole> result = new HashSet<>();
+ for (String role : getList(KRaftConfigs.PROCESS_ROLES_CONFIG)) {
+ switch (role) {
+ case "broker" -> result.add(ProcessRole.BrokerRole);
+ case "controller" -> result.add(ProcessRole.ControllerRole);
+ default -> throw new ConfigException("Unknown process role '" + role +
+ "' (only 'broker' and 'controller' are allowed roles)");
+ }
+ }
+ value = Collections.unmodifiableSet(result);
+ processRoles = value;
+ }
+ return value;
+ }
+
+ public boolean isKRaftCombinedMode(Set<ProcessRole> processRoles) {
+ return processRoles.equals(Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole));
+ }
+
+ public String metadataLogDir() {
+ String dir = getString(MetadataLogConfig.METADATA_LOG_DIR_CONFIG);
+ return dir != null ? dir : logDirs().get(0);
+ }
+
+ public long serverMaxStartupTimeMs() {
+ return getLong(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG);
+ }
+
+ public Integer messageMaxBytes() {
+ return getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
+ }
+
+ public int getNumReplicaAlterLogDirsThreads() {
+ Integer numThreads = getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG);
+ return numThreads != null ? numThreads : logDirs().size();
+ }
+
+ // ********* Metadata Configuration **********
+
+ public long metadataSnapshotMaxNewRecordBytes() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG);
+ }
+
+ public long metadataSnapshotMaxIntervalMs() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG);
+ }
+
+ public OptionalLong metadataMaxIdleIntervalNs() {
+ long value = TimeUnit.NANOSECONDS.convert(
+ getInt(MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG).longValue(),
+ TimeUnit.MILLISECONDS);
+ return value > 0 ? OptionalLong.of(value) : OptionalLong.empty();
+ }
+
+ // ********* Rack Configuration **********
+
+ public Optional<String> replicaSelectorClassName() {
+ return Optional.ofNullable(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG));
+ }
+
+ // ********* Replication Configuration **********
+
+ public long replicaLagTimeMaxMs() {
+ return getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG);
+ }
+
+ public int replicaSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int replicaSocketReceiveBufferBytes() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_RECEIVE_BUFFER_BYTES_CONFIG);
+ }
+
+ public int replicaFetchMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchWaitMaxMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG);
+ }
+
+ public int replicaFetchMinBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MIN_BYTES_CONFIG);
+ }
+
+ public int replicaFetchResponseMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchBackoffMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_BACKOFF_MS_CONFIG);
+ }
+
+ public long replicaHighWatermarkCheckpointIntervalMs() {
+ return getLong(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG);
+ }
+
+ public int fetchPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int producerPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.PRODUCER_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int deleteRecordsPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.DELETE_RECORDS_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public boolean autoLeaderRebalanceEnable() {
+ return getBoolean(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG);
+ }
+
+ public long leaderImbalanceCheckIntervalSeconds() {
+ return getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG);
+ }
+
+ public long uncleanLeaderElectionCheckIntervalMs() {
+ return getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG);
+ }
+
+ public Boolean uncleanLeaderElectionEnable() {
+ return getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG);
+ }
+
+ public Boolean followerFetchLastTieredOffsetEnable() {
+ return getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
+ }
+
+ // ********* Controlled Shutdown Configuration **********
+
+ public boolean controlledShutdownEnable() {
+ return getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG);
+ }
+
+ // ********* Group Coordinator Configuration **********
+
+ @SuppressWarnings("removal")
+ public Set<GroupType> groupCoordinatorRebalanceProtocols() {
+ return getList(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)
+ .stream()
+ .map(s -> GroupType.valueOf(s.toUpperCase(Locale.ROOT)))
+ .collect(Collectors.toUnmodifiableSet());
+ }
+
+ @SuppressWarnings("removal")
+ protected void validateGroupCoordinatorRebalanceProtocols() {
+ Set<GroupType> protocols = groupCoordinatorRebalanceProtocols();
+
+ if (!protocols.contains(GroupType.CLASSIC)) {
+ throw new ConfigException("Disabling the '" + GroupType.CLASSIC + "' protocol is not supported.");
+ }
+
+ if (doLog && protocols.contains(GroupType.SHARE)) {
+ log.warn("'{}' in `{}` is deprecated. " +
+ "Share groups are controlled by the 'share.version' feature. " +
+ "This config will be removed in Kafka 5.0.",
+ GroupType.SHARE,
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG);
+ }
+
+ if (doLog && originals().containsKey(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)) {
+ Set<GroupType> defaultProtocols = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT
+ .stream()
+ .map(s -> GroupType.valueOf(s.toUpperCase(Locale.ROOT)))
+ .collect(Collectors.toUnmodifiableSet());
+
+ Set<GroupType> missingProtocols = new HashSet<>(defaultProtocols);
+ missingProtocols.removeAll(protocols);
+
+ if (!missingProtocols.isEmpty()) {
+ log.warn("The config `{}` is deprecated and will be removed in Kafka 5.0. " +
+ "The following protocol(s) are currently disabled: {}. " +
+ "In Kafka 5.0, all protocols will always be enabled and controlled solely by feature versions " +
+ "(group.version, streams.version, share.version) via kafka-features.sh. " +
+ "Please remove the configuration, which will restore all protocols to the default enabled state, to prepare for the upgrade.",
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG,
+ missingProtocols.stream().map(GroupType::toString).collect(Collectors.joining(", ")));
+ } else {
+ log.warn("The config `{}` is deprecated and will be removed in Kafka 5.0. " +
+ "Please remove the configuration to prepare for the upgrade.",
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG);
+ }
+ }
+ }
+
+ // ********* Metric Configuration **********
+
+ public int metricNumSamples() {
+ return getInt(MetricConfigs.METRIC_NUM_SAMPLES_CONFIG);
+ }
+
+ public long metricSampleWindowMs() {
+ return getLong(MetricConfigs.METRIC_SAMPLE_WINDOW_MS_CONFIG);
+ }
+
+ public String metricRecordingLevel() {
+ return getString(MetricConfigs.METRIC_RECORDING_LEVEL_CONFIG);
+ }
+
+ // ********* Kafka Client Telemetry Metrics Configuration **********
+
+ public int clientTelemetryMaxBytes() {
+ return getInt(MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_CONFIG);
+ }
+
+ // ********* SSL/SASL Configuration **********
+ // Security configs may be overridden for listeners, so it is not safe to use the base values.
+ // Hence the base SSL/SASL configs are not fields of KafkaConfig, listener configs should be
+ // retrieved using KafkaConfig#valuesWithPrefixOverride
+
+ @SuppressWarnings("unchecked")
+ public Set<String> saslEnabledMechanisms(ListenerName listenerName) {
Review Comment:
Great catch! I did one more sweep at this is the only originally private method which was moved across.
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3918828693
##########
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java:
##########
@@ -45,7 +45,7 @@
*/
public class AbstractConfig {
- private static final Logger log = LoggerFactory.getLogger(AbstractConfig.class);
+ public static final Logger log = LoggerFactory.getLogger(AbstractConfig.class);
Review Comment:
> it to protected
This is a bit of a grey area. We haven't reached consensus on whether protected members count as public API when the public class is meant to be extended. To sidestep that, I'd prefer that AbstractKafkaConfig has its own LOGGER
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3952887754
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -125,7 +121,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
// Cache the current config to avoid acquiring read lock to access from dynamicConfig
@volatile private var currentConfig = this
- val processRoles: Set[ProcessRole] = parseProcessRoles()
+ val processRoles: Set[ProcessRole] = parseProcessRoles().asScala.toSet
Review Comment:
It seems `processRoles` is not a hot method either, so we can move it to `AbstractKafkaConfig`. For example: `Set<ProcessRole> processRoles`
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3968287898
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -552,7 +411,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
validateControllerQuorumVotersMustContainNodeIdForKRaftController()
validateAdvertisedControllerListenersNonEmptyForKRaftController()
validateControllerListenerNamesMustAppearInListenersForKRaftController()
- } else if (isKRaftCombinedMode) {
+ } else if (isKRaftCombinedMode(processRoles)) {
Review Comment:
It was also used in a test in KafkaConfigTest, but I agree with you. Inlined it in both locations!
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3918885251
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -635,4 +642,295 @@ public Map<String, Object> extractGroupConfigMap() {
);
return defaults;
}
+
+ // ********* General Configuration **********
+
+ public int brokerSessionTimeoutMs() {
+ return getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG);
+ }
+
+ public long controllerPerformanceSamplePeriodMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS);
+ }
+
+ public long controllerPerformanceAlwaysLogThresholdMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS);
+ }
+
+ public Set<ProcessRole> parseProcessRoles() {
+ List<String> roles = getList(KRaftConfigs.PROCESS_ROLES_CONFIG);
+ Set<ProcessRole> result = new HashSet<>();
+ for (String role : roles) {
+ switch (role) {
+ case "broker" -> result.add(ProcessRole.BrokerRole);
+ case "controller" -> result.add(ProcessRole.ControllerRole);
+ default -> throw new ConfigException("Unknown process role '" + role +
+ "' (only 'broker' and 'controller' are allowed roles)");
+ }
+ }
+ return Collections.unmodifiableSet(result);
+ }
+
+ public boolean isKRaftCombinedMode(Set<ProcessRole> processRoles) {
+ return processRoles.equals(Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole));
+ }
+
+ public String metadataLogDir() {
+ String dir = getString(MetadataLogConfig.METADATA_LOG_DIR_CONFIG);
+ return dir != null ? dir : logDirs().get(0);
+ }
+
+ public long serverMaxStartupTimeMs() {
+ return getLong(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG);
+ }
+
+ public Integer messageMaxBytes() {
+ return getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
+ }
+
+ public int getNumReplicaAlterLogDirsThreads() {
+ Integer numThreads = getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG);
+ return numThreads != null ? numThreads : logDirs().size();
+ }
+
+ // ********* Metadata Configuration **********
+
+ public long metadataSnapshotMaxNewRecordBytes() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG);
+ }
+
+ public long metadataSnapshotMaxIntervalMs() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG);
+ }
+
+ public Optional<Long> metadataMaxIdleIntervalNs() {
+ long value = TimeUnit.NANOSECONDS.convert(
+ getInt(MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG).longValue(),
+ TimeUnit.MILLISECONDS);
+ return value > 0 ? Optional.of(value) : Optional.empty();
+ }
+
+ // ********* Rack Configuration **********
+
+ public Optional<String> replicaSelectorClassName() {
+ return Optional.ofNullable(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG));
+ }
+
+ // ********* Replication Configuration **********
+
+ public int controllerSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.CONTROLLER_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int defaultReplicationFactor() {
+ return getInt(ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG);
+ }
+
+ public long replicaLagTimeMaxMs() {
+ return getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG);
+ }
+
+ public int replicaSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int replicaSocketReceiveBufferBytes() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_RECEIVE_BUFFER_BYTES_CONFIG);
+ }
+
+ public int replicaFetchMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchWaitMaxMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG);
+ }
+
+ public int replicaFetchMinBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MIN_BYTES_CONFIG);
+ }
+
+ public int replicaFetchResponseMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchBackoffMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_BACKOFF_MS_CONFIG);
+ }
+
+ public long replicaHighWatermarkCheckpointIntervalMs() {
+ return getLong(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG);
+ }
+
+ public int fetchPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int producerPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.PRODUCER_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int deleteRecordsPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.DELETE_RECORDS_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public boolean autoLeaderRebalanceEnable() {
+ return getBoolean(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG);
+ }
+
+ public long leaderImbalanceCheckIntervalSeconds() {
+ return getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG);
+ }
+
+ public long uncleanLeaderElectionCheckIntervalMs() {
+ return getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG);
+ }
+
+ public Boolean uncleanLeaderElectionEnable() {
+ return getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG);
+ }
+
+ public Boolean followerFetchLastTieredOffsetEnable() {
+ return getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
+ }
+
+ // ********* Controlled Shutdown Configuration **********
+
+ public boolean controlledShutdownEnable() {
+ return getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG);
+ }
+
+ // ********* Group Coordinator Configuration **********
+
+ @SuppressWarnings("removal")
+ public Set<GroupType> groupCoordinatorRebalanceProtocols() {
Review Comment:
Could we move the validation and warning into validateValues, and have groupCoordinatorRebalanceProtocols compute the value on demand?
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3044242257
##########
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java:
##########
@@ -45,7 +45,7 @@
*/
public class AbstractConfig {
- private static final Logger log = LoggerFactory.getLogger(AbstractConfig.class);
+ public static final Logger log = LoggerFactory.getLogger(AbstractConfig.class);
Review Comment:
Fair suggestion, we only need this up until we get rid of the configurations which trigger the warnings, but I will swap it to protected
--
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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "mingyen066 (via GitHub)" <gi...@apache.org>.
mingyen066 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r3971118399
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -90,9 +99,12 @@ public abstract class AbstractKafkaConfig extends AbstractConfig {
));
private volatile QuotaConfig quotaConfig;
+ private volatile Set<ProcessRole> processRoles;
Review Comment:
Could these two fields be removed as well? The `processRoles` cache was deferred as discussed above, and `quotaConfig()` was already removed from trunk in #23362, so this line looks like a rebase leftover. Nothing reads or writes either field 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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21943:
URL: https://github.com/apache/kafka/pull/21943#discussion_r4081688232
##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -670,4 +679,285 @@ public Map<String, Object> extractGroupConfigMap(GroupCoordinatorConfig groupCoo
);
return defaults;
}
+
+ // ********* General Configuration **********
+
+ public int brokerSessionTimeoutMs() {
+ return getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG);
+ }
+
+ public long controllerPerformanceSamplePeriodMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS);
+ }
+
+ public long controllerPerformanceAlwaysLogThresholdMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS);
+ }
+
+ public Set<ProcessRole> processRoles() {
Review Comment:
opened https://issues.apache.org/jira/browse/KAFKA-21150 to refactor 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-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #21943:
URL: https://github.com/apache/kafka/pull/21943
--
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