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