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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
