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/02/23 15:33:22 UTC

[PR] KAFKA-15853: Move static methods and SocketServer configs to AbstractKafkaConfig [kafka]

clolov opened a new pull request, #21548:
URL: https://github.com/apache/kafka/pull/21548

   This is an attempt to propose a gradual migration of KafkaConfig to Java for easier reviews.
   
   In this draft I moved the static methods to AbstractKafkaConfig and the methods used to access SocketServer properties.
   
   If this seems like a sensible approach there will be follow-up PRs.


-- 
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: Move static methods and SocketServer configs to AbstractKafkaConfig [kafka]

Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.

UladzislauBlok commented on PR #21548:
URL: https://github.com/apache/kafka/pull/21548#issuecomment-3958379827

   @clolov Hello. I will re-assign ticket to you. Thanks for picking this up. That's the big one :)


-- 
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: [1/N] Move static methods, SocketServer and Log configs to AbstractKafkaConfig [kafka]

Posted by "clolov (via GitHub)" <gi...@apache.org>.

clolov commented on code in PR #21548:
URL: https://github.com/apache/kafka/pull/21548#discussion_r2987671650


##########
server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java:
##########
@@ -350,4 +356,279 @@ private static String parseListenerName(String connectionString) {
      * @param reconfigurable the component to unregister
      */
     public abstract void removeReconfigurable(Reconfigurable reconfigurable);
+
+    /**
+     * Determines whether a config entry might be sensitive based on its type.
+     * If the type cannot be determined, the config is treated as sensitive
+     * to be safe.
+     *
+     * @param configType the config entry type, or empty if unknown
+     * @return true if the config might be sensitive
+     */
+    public static boolean maybeSensitive(Optional<ConfigDef.Type> configType) {
+        return configType.isEmpty()
+                || configType.get() == ConfigDef.Type.PASSWORD;
+    }
+
+    /**
+     * Looks up the type for a config key by name directly from
+     * {@link #CONFIG_DEF}.
+     *
+     * @param name the config key name
+     * @return the type if found, or empty
+     */
+    public static Optional<ConfigDef.Type> configDefTypeOf(String name) {
+        return Optional.ofNullable(CONFIG_DEF.configKeys().get(name))
+                .map(key -> key.type);
+    }
+
+    /**
+     * Resolves the {@link ConfigDef.Type} for a given config name.
+     * First tries an exact match, then falls back to checking broker
+     * config synonyms.
+     *
+     * @param configName the config name to look up
+     * @return the type if found, or empty
+     */
+    public static Optional<ConfigDef.Type> configType(String configName) {
+        return configDefTypeOf(configName)
+                .or(() -> Optional.ofNullable(DynamicConfig.Broker.configKeys().get(configName))
+                        .map(key -> key.type))
+                .or(() -> DynamicBrokerConfig.brokerConfigSynonyms(configName, true)
+                        .stream()
+                        .map(AbstractKafkaConfig::configDefTypeOf)
+                        .flatMap(Optional::stream)
+                        .findFirst());
+    }
+
+    /**
+     * Returns the loggable form of a config value. Sensitive values
+     * are replaced with {@link Password#HIDDEN}.
+     *
+     * @param resourceType the config resource type
+     * @param name         the config name
+     * @param value        the config value
+     * @return the value suitable for logging
+     */
+    public static String loggableValue(ConfigResource.Type resourceType,
+                                       String name,
+                                       String value) {
+        boolean sensitive = switch (resourceType) {
+            case BROKER -> maybeSensitive(configType(name));
+            case TOPIC -> maybeSensitive(LogConfig.configType(name));
+            case GROUP -> maybeSensitive(GroupConfig.configType(name));
+            case BROKER_LOGGER, CLIENT_METRICS -> false;
+            default -> true;
+        };
+        return sensitive ? Password.HIDDEN : value;
+    }
+
+    // ********* Socket Server Configuration **********
+
+    public int socketSendBufferBytes() {
+        return getInt(SocketServerConfigs.SOCKET_SEND_BUFFER_BYTES_CONFIG);
+    }
+
+    public int socketReceiveBufferBytes() {
+        return getInt(SocketServerConfigs.SOCKET_RECEIVE_BUFFER_BYTES_CONFIG);
+    }
+
+    public int socketRequestMaxBytes() {
+        return getInt(SocketServerConfigs.SOCKET_REQUEST_MAX_BYTES_CONFIG);
+    }
+
+    public int socketListenBacklogSize() {
+        return getInt(SocketServerConfigs.SOCKET_LISTEN_BACKLOG_SIZE_CONFIG);
+    }
+
+    public int maxConnectionsPerIp() {
+        return getInt(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_CONFIG);
+    }
+
+    public Map<String, Integer> maxConnectionsPerIpOverrides() {
+        return getMap(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_OVERRIDES_CONFIG,
+                getString(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_OVERRIDES_CONFIG))
+                .entrySet()
+                .stream()
+                .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, e -> Integer.parseInt(e.getValue())));
+    }
+
+    public int maxConnections() {
+        return getInt(SocketServerConfigs.MAX_CONNECTIONS_CONFIG);
+    }
+
+    public int maxConnectionCreationRate() {
+        return getInt(SocketServerConfigs.MAX_CONNECTION_CREATION_RATE_CONFIG);
+    }
+
+    public long connectionsMaxIdleMs() {
+        return getLong(SocketServerConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG);
+    }
+
+    public int failedAuthenticationDelayMs() {
+        return getInt(SocketServerConfigs.FAILED_AUTHENTICATION_DELAY_MS_CONFIG);
+    }
+
+    public int queuedMaxRequests() {
+        return getInt(SocketServerConfigs.QUEUED_MAX_REQUESTS_CONFIG);
+    }
+
+    public long queuedMaxBytes() {
+        return getLong(SocketServerConfigs.QUEUED_MAX_BYTES_CONFIG);
+    }
+
+    public int numNetworkThreads() {
+        return getInt(SocketServerConfigs.NUM_NETWORK_THREADS_CONFIG);
+    }
+
+    public long connectionSetupTimeoutMs() {
+        return getLong(ServerConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG);
+    }
+
+    public long connectionSetupTimeoutMaxMs() {
+        return getLong(ServerConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG);
+    }
+
+    // ********* Log Configuration **********
+
+    public boolean autoCreateTopicsEnable() {
+        return getBoolean(ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG);
+    }
+
+    public int numPartitions() {
+        return getInt(ServerLogConfigs.NUM_PARTITIONS_CONFIG);
+    }
+
+    public Integer logSegmentBytes() {
+        return getInt(ServerLogConfigs.LOG_SEGMENT_BYTES_CONFIG);
+    }
+
+    public Long logFlushIntervalMessages() {
+        return getLong(ServerLogConfigs.LOG_FLUSH_INTERVAL_MESSAGES_CONFIG);
+    }
+
+    public int logCleanerThreads() {
+        return getInt(CleanerConfig.LOG_CLEANER_THREADS_PROP);
+    }
+
+    public long logFlushSchedulerIntervalMs() {
+        return getLong(ServerLogConfigs.LOG_FLUSH_SCHEDULER_INTERVAL_MS_CONFIG);
+    }
+
+    public long logFlushOffsetCheckpointIntervalMs() {
+        return getInt(ServerLogConfigs.LOG_FLUSH_OFFSET_CHECKPOINT_INTERVAL_MS_CONFIG).longValue();
+    }
+
+    public long logFlushStartOffsetCheckpointIntervalMs() {
+        return getInt(ServerLogConfigs.LOG_FLUSH_START_OFFSET_CHECKPOINT_INTERVAL_MS_CONFIG).longValue();
+    }
+
+    public long logCleanupIntervalMs() {
+        return getLong(ServerLogConfigs.LOG_CLEANUP_INTERVAL_MS_CONFIG);
+    }
+
+    public List<String> logCleanupPolicy() {
+        return getList(ServerLogConfigs.LOG_CLEANUP_POLICY_CONFIG);
+    }
+
+    public Long logRetentionBytes() {
+        return getLong(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG);
+    }
+
+    public long logCleanerDedupeBufferSize() {
+        return getLong(CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_SIZE_PROP);
+    }
+
+    public Long logCleanerDeleteRetentionMs() {
+        return getLong(CleanerConfig.LOG_CLEANER_DELETE_RETENTION_MS_PROP);
+    }
+
+    public Long logCleanerMinCompactionLagMs() {
+        return getLong(CleanerConfig.LOG_CLEANER_MIN_COMPACTION_LAG_MS_PROP);
+    }
+
+    public Long logCleanerMaxCompactionLagMs() {
+        return getLong(CleanerConfig.LOG_CLEANER_MAX_COMPACTION_LAG_MS_PROP);
+    }
+
+    public Double logCleanerMinCleanRatio() {
+        return getDouble(CleanerConfig.LOG_CLEANER_MIN_CLEAN_RATIO_PROP);
+    }
+
+    public Integer logIndexSizeMaxBytes() {
+        return getInt(ServerLogConfigs.LOG_INDEX_SIZE_MAX_BYTES_CONFIG);
+    }
+
+    public Integer logIndexIntervalBytes() {
+        return getInt(ServerLogConfigs.LOG_INDEX_INTERVAL_BYTES_CONFIG);
+    }
+
+    public Long logDeleteDelayMs() {
+        return getLong(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG);
+    }
+
+    public Long logRollTimeMillis() {
+        Long millis = getLong(ServerLogConfigs.LOG_ROLL_TIME_MILLIS_CONFIG);
+        if (millis != null) return millis;
+        return 60L * 60L * 1000L * getInt(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG);
+    }
+
+    public Long logRollTimeJitterMillis() {
+        Long millis = getLong(ServerLogConfigs.LOG_ROLL_TIME_JITTER_MILLIS_CONFIG);
+        if (millis != null) return millis;
+        return 60L * 60L * 1000L * getInt(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG);
+    }
+
+    public Long logFlushIntervalMs() {
+        Long millis = getLong(ServerLogConfigs.LOG_FLUSH_INTERVAL_MS_CONFIG);
+        if (millis != null) return millis;
+        return getLong(ServerLogConfigs.LOG_FLUSH_SCHEDULER_INTERVAL_MS_CONFIG);
+    }
+
+    public Integer minInSyncReplicas() {
+        return getInt(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_CONFIG);
+    }
+
+    public Boolean logPreAllocateEnable() {
+        return getBoolean(ServerLogConfigs.LOG_PRE_ALLOCATE_CONFIG);
+    }
+
+    public long logInitialTaskDelayMs() {
+        Long millis = getLong(ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_CONFIG);
+        if (millis != null) return millis;
+        return ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_DEFAULT;
+    }
+
+    public TimestampType logMessageTimestampType() {
+        return TimestampType.forName(getString(ServerLogConfigs.LOG_MESSAGE_TIMESTAMP_TYPE_CONFIG));
+    }
+
+    public long logMessageTimestampBeforeMaxMs() {
+        return getLong(ServerLogConfigs.LOG_MESSAGE_TIMESTAMP_BEFORE_MAX_MS_CONFIG);
+    }
+
+    public long logMessageTimestampAfterMaxMs() {
+        return getLong(ServerLogConfigs.LOG_MESSAGE_TIMESTAMP_AFTER_MAX_MS_CONFIG);
+    }
+
+    public long logDirFailureTimeoutMs() {
+        return getLong(ServerLogConfigs.LOG_DIR_FAILURE_TIMEOUT_MS_CONFIG);
+    }
+
+    public Long logRetentionTimeMillis() {

Review Comment:
   For this one, I have changed to use TimeUnit, but the last line needs to be `return millis < 0 ? Long.valueOf(-1) : millis;`. Otherwise spotbugs complains with
   ```
   Boxed value is unboxed and then immediately reboxed in org.apache.kafka.server.config.AbstractKafkaConfig.logRetentionTimeMillis()
   ```
   This is one of those boxing/unboxing issues which ought to disappear once we move fully 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