Current state: [One of "Under Discussion", "Accepted", "Rejected"]
Discussion thread: here
JIRA: here
The configuration can be applied at the topic, broker, or cluster level. However, for some configurations, our primary concern is the behavior of topics, or ensuring that topics operate consistently across the cluster, rather than expecting the same topic to behave differently on different brokers. For example, if there are two brokers with different settings for min.insync.replicas, moving a leader to a different broker would change its behavior with respect to the min.insync.replicas semantics.
This KIP aims to enforce that the proposed configurations have the same cluster-level settings and disallow them from being set at the broker level, ensuring consistent behavior across the cluster.
On the cluster's first startup, initialize the proposed configurations at the cluster level. Since dynamic configs have higher priority, any corresponding static broker configs will not take effect.
Disallow setting the proposed configurations at the broker level and removal of proposed configurations at the cluster level. The corresponding errors will be thrown if such requests are received.
When using the updateFeatures API to upgrade to a MetadataVersion that includes this KIP change:
Specifically, introduce a new enum GuardedBrokerConfig that includes the proposed configurations and helper methods to keep the code concise and extensible.
public enum GuardedBrokerConfig {
MIN_IN_SYNC_REPLICAS(MIN_IN_SYNC_REPLICAS_CONFIG, ConfigDef.Type.INT),
UNCLEAN_LEADER_ELECTION_ENABLE(UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, ConfigDef.Type.BOOLEAN),
MESSAGE_MAX_BYTES(MESSAGE_MAX_BYTES_CONFIG, ConfigDef.Type.INT),
LOG_MESSAGE_TIMESTAMP_TYPE(LOG_MESSAGE_TIMESTAMP_TYPE_CONFIG, ConfigDef.Type.STRING),
LOG_CLEANUP_POLICY(LOG_CLEANUP_POLICY_CONFIG, ConfigDef.Type.STRING);
private final String configName;
private final ConfigDef.Type type;
private static final Map<String, GuardedBrokerConfig> NAME_TO_CONFIG = new HashMap<>();
static {
for (GuardedBrokerConfig config : GuardedBrokerConfig.values()) {
NAME_TO_CONFIG.put(config.configName, config);
}
}
GuardedBrokerConfig(String name, ConfigDef.Type type) {
this.configName = name;
this.type = type;
}
// ... remaining helper methods ... //
} |
For the incrementalAlterConfigs and AlterConfigs APIs, any disallowed requests will be rejected and immediately throw an error. This may break user applications.
The typical suite of unit/integration tests will be added.
1. Disallow alter API requests with warnings and ignore setting
Instead of immediately rejecting disallowed requests with an error, only log a warning and ignore the setting. This would avoid breaking user applications until the next major release.
Reason for Rejection: Users may easily overlook warnings and not realize their configurations have not taken effect, which could lead to even more severe operational issues later.
2. Each proposed config with its own logic
Instead of introducing a new enum structure, handle each configuration individually with its own logic (similar to how min.insync.replicas is currently implemented, see https://github.com/apache/kafka/pull/17952),
Reason for Rejection: This approach is harder to extend and more error-prone. Having a centralized structure provides a cleaner and more extensible solution.