KAFKA-18948 Move DynamicLogConfig to server module#22353
Conversation
unknowntpo
force-pushed
the
move-dynamic-log-config
branch
2 times, most recently
from
16d254e to
ce74012
Compare
github-actions
Bot
added
the
storage
m1a2st
left a comment
m1a2st
left a comment
There was a problem hiding this comment.
Thanks @unknowntpo for this patch!
| import java.util.stream.Collectors; | ||
| import java.util.stream.Stream; | ||
|
|
||
| public class DynamicLogConfig implements BrokerReconfigurable { |
There was a problem hiding this comment.
We should add new test for this new config class.
There was a problem hiding this comment.
Okay, DynamicLogConfigTest are added, please take a look, thanks.
unknowntpo
force-pushed
the
move-dynamic-log-config
branch
2 times, most recently
from
3732ab0 to
ce74012
Compare
chia7712
changed the title
| logManager.brokerConfigUpdated(); | ||
| for (UnifiedLog unifiedLog : logManager.allLogs()) { | ||
| Map<String, Object> props = new HashMap<>(newBrokerDefaults); | ||
| unifiedLog.config().originals().entrySet().stream() |
There was a problem hiding this comment.
We could call forEach directly
unifiedLog.config().originals().forEach((k, v) -> {
if (unifiedLog.config().overriddenConfigs.contains(k)) {
props.put(k, v);
}
});
| private void validateLogLocalRetentionMs(AbstractKafkaConfig config) { | ||
| long logRetentionMs = config.logRetentionTimeMillis(); | ||
| long logLocalRetentionMs = config.getLong(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP); | ||
| if (logRetentionMs != LogConfig.NO_RETENTION_LIMIT && logLocalRetentionMs != LogConfig.DEFAULT_LOCAL_RETENTION_MS) { |
There was a problem hiding this comment.
I also replaced raw -1 / -2 retention sentinels with LogConfig constants. This keeps DynamicLogConfig aligned with LogConfig semantics and makes the validation easier to read without changing behavior.
unknowntpo
force-pushed
the
move-dynamic-log-config
branch
from
8c6f7a1 to
d2c8e74
Compare
| validateCordonedLogDirs(kafkaConfig); | ||
| } | ||
|
|
||
| private AbstractKafkaConfig requireKafkaConfig(AbstractConfig config) { |
There was a problem hiding this comment.
Maybe we could add generics to BrokerReconfigurable? For example, BrokerReconfigurable<T extends AbstractConfig>. This would allow both the storage and server modules to access the specific config types they need
There was a problem hiding this comment.
Good point. I don't like the cast either. A generic BrokerReconfigurable<T extends AbstractConfig> sounds like the right direction, but it touches the interface and all implementers, so I will keep this PR focused and
handle that in a follow-up PR.
3 participants
Summary
Move
DynamicLogConfigfromcoreto theservermodule as a JavaBrokerReconfigurable.This also moves
extractLogConfigMaptoAbstractKafkaConfig, so logreconfiguration no longer depends directly on
kafka.server.KafkaConfig.Note: this branch temporarily includes the generic
BrokerReconfigurablechange from #22409 soDynamicLogConfigcan useAbstractKafkaConfigdirectly. Drop that commit after #22409 is merged.Testing
./gradlew :core:test --tests kafka.server.DynamicBrokerConfigTestReviewers: Ken Huang s7133700@gmail.com, Chia-Ping Tsai
chia7712@gmail.com