Jackie-Jiang commented on code in PR #19686:
URL: https://github.com/apache/pinot/pull/19686#discussion_r4251535559
##########
pinot-common/src/main/java/org/apache/pinot/common/config/DefaultClusterConfigChangeHandler.java:
##########
@@ -19,50 +19,129 @@
package org.apache.pinot.common.config;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
import org.apache.helix.NotificationContext;
import org.apache.helix.api.listeners.BatchMode;
import org.apache.helix.api.listeners.ClusterConfigChangeListener;
import org.apache.helix.model.ClusterConfig;
import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
import org.apache.pinot.spi.config.provider.PinotClusterConfigProvider;
+import org.apache.pinot.spi.env.PinotConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/// Bridges Helix cluster config callbacks to
[PinotClusterConfigChangeListener]s. All access is serialized on this
-/// instance, so a listener sees the snapshot handed to it at registration and
every later change in order.
+/// Bridges Helix cluster config callbacks to
[PinotClusterConfigChangeListener]s. Instance configs captured at
+/// construction override cluster configs in the effective view exposed to
listeners and [#getClusterConfigs()]. All
+/// access is serialized on this instance, so a listener sees the snapshot
handed to it at registration and every later
+/// change in order.
@BatchMode(enabled = false)
public class DefaultClusterConfigChangeHandler implements
ClusterConfigChangeListener, PinotClusterConfigProvider {
private static final Logger LOGGER =
LoggerFactory.getLogger(DefaultClusterConfigChangeHandler.class);
+ private static final Comparator<String> CONFIG_KEY_COMPARATOR =
+
Comparator.comparing(DefaultClusterConfigChangeHandler::normalizeConfigKey);
private final List<PinotClusterConfigChangeListener> _listeners = new
ArrayList<>();
- private Map<String, String> _clusterConfigs = Map.of();
+ private final Map<String, String> _instanceConfigs;
+ private Map<String, String> _effectiveConfigs;
+
+ public DefaultClusterConfigChangeHandler() {
+ this(Map.of());
+ }
+
+ /// Snapshots the instance config at construction time. Callers must
construct the handler before folding cluster
+ /// configs into the instance config.
+ /// @param instanceConfig instance config to snapshot
+ public DefaultClusterConfigChangeHandler(PinotConfiguration instanceConfig) {
+ this(snapshot(instanceConfig));
+ }
+
+ DefaultClusterConfigChangeHandler(Map<String, String> instanceConfigs) {
+ _instanceConfigs = immutableConfigMap(instanceConfigs);
+ _effectiveConfigs = _instanceConfigs;
+ }
@Override
public synchronized void onClusterConfigChange(ClusterConfig clusterConfig,
NotificationContext context) {
Map<String, String> clusterConfigs =
copyWithoutNullValues(clusterConfig.getRecord().getSimpleFields());
- Set<String> changedConfigs = getChangedProperties(_clusterConfigs,
clusterConfigs);
+ Map<String, String> effectiveConfigs = getEffectiveConfigs(clusterConfigs);
+ Set<String> changedConfigs =
getChangedEffectiveProperties(_effectiveConfigs, effectiveConfigs);
LOGGER.info("Cluster configs changed: {}", changedConfigs);
- _clusterConfigs = clusterConfigs;
+ _effectiveConfigs = effectiveConfigs;
for (PinotClusterConfigChangeListener listener : _listeners) {
- listener.onChange(changedConfigs, clusterConfigs);
+ listener.onChange(changedConfigs, effectiveConfigs);
}
}
@Override
public synchronized Map<String, String> getClusterConfigs() {
- return _clusterConfigs;
+ return _effectiveConfigs;
}
@Override
public synchronized boolean
registerClusterConfigChangeListener(PinotClusterConfigChangeListener listener) {
LOGGER.info("Registering cluster config change listener: {}",
listener.getClass().getName());
_listeners.add(listener);
// Treat every key as newly added so that the listener picks up the
current values
- listener.onChange(_clusterConfigs.keySet(), _clusterConfigs);
+ listener.onChange(_effectiveConfigs.keySet(), _effectiveConfigs);
Review Comment:
[P1] Seed the complete effective configuration before replaying it
With instance `pinot.server.consumption.rate.limit=0` and cluster
`pinot.server.consumption.rate.limit.bytes=1000`, `BaseServerStarter` first
creates the correct byte limiter from its merged configuration. This new
registration callback contains only instance settings, so
`ServerRateLimitConfigChangeListener` defaults the missing byte setting to zero
and replaces that limiter with `NOOP_RATE_LIMITER`.
Helix connects at `BaseServerStarter:807`, before the cluster-config
listener is registered at line 825. A consuming segment constructed in that
window caches the no-op limiter in the final
`RealtimeSegmentDataManager._serverRateLimiter` field (assignment at line
1940). The first cluster callback creates a new real limiter, but the existing
consumer keeps its no-op reference and bypasses the server cap for the
segment's lifetime.
An exact-source reproduction shows real limiter → no-op → new real limiter
on this head; the base preserves the original real limiter throughout. Please
initialize the complete effective snapshot before replaying it, or ensure
cluster configuration is applied before consumer state transitions can create
segments. Add a startup-order regression test covering a rate limit split
between instance and cluster configuration.
##########
pinot-common/src/main/java/org/apache/pinot/common/config/DefaultClusterConfigChangeHandler.java:
##########
@@ -19,50 +19,129 @@
package org.apache.pinot.common.config;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
import org.apache.helix.NotificationContext;
import org.apache.helix.api.listeners.BatchMode;
import org.apache.helix.api.listeners.ClusterConfigChangeListener;
import org.apache.helix.model.ClusterConfig;
import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
import org.apache.pinot.spi.config.provider.PinotClusterConfigProvider;
+import org.apache.pinot.spi.env.PinotConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/// Bridges Helix cluster config callbacks to
[PinotClusterConfigChangeListener]s. All access is serialized on this
-/// instance, so a listener sees the snapshot handed to it at registration and
every later change in order.
+/// Bridges Helix cluster config callbacks to
[PinotClusterConfigChangeListener]s. Instance configs captured at
+/// construction override cluster configs in the effective view exposed to
listeners and [#getClusterConfigs()]. All
+/// access is serialized on this instance, so a listener sees the snapshot
handed to it at registration and every later
+/// change in order.
@BatchMode(enabled = false)
public class DefaultClusterConfigChangeHandler implements
ClusterConfigChangeListener, PinotClusterConfigProvider {
private static final Logger LOGGER =
LoggerFactory.getLogger(DefaultClusterConfigChangeHandler.class);
+ private static final Comparator<String> CONFIG_KEY_COMPARATOR =
+
Comparator.comparing(DefaultClusterConfigChangeHandler::normalizeConfigKey);
private final List<PinotClusterConfigChangeListener> _listeners = new
ArrayList<>();
- private Map<String, String> _clusterConfigs = Map.of();
+ private final Map<String, String> _instanceConfigs;
+ private Map<String, String> _effectiveConfigs;
+
+ public DefaultClusterConfigChangeHandler() {
+ this(Map.of());
+ }
+
+ /// Snapshots the instance config at construction time. Callers must
construct the handler before folding cluster
+ /// configs into the instance config.
+ /// @param instanceConfig instance config to snapshot
+ public DefaultClusterConfigChangeHandler(PinotConfiguration instanceConfig) {
+ this(snapshot(instanceConfig));
+ }
+
+ DefaultClusterConfigChangeHandler(Map<String, String> instanceConfigs) {
+ _instanceConfigs = immutableConfigMap(instanceConfigs);
+ _effectiveConfigs = _instanceConfigs;
+ }
@Override
public synchronized void onClusterConfigChange(ClusterConfig clusterConfig,
NotificationContext context) {
Map<String, String> clusterConfigs =
copyWithoutNullValues(clusterConfig.getRecord().getSimpleFields());
- Set<String> changedConfigs = getChangedProperties(_clusterConfigs,
clusterConfigs);
+ Map<String, String> effectiveConfigs = getEffectiveConfigs(clusterConfigs);
+ Set<String> changedConfigs =
getChangedEffectiveProperties(_effectiveConfigs, effectiveConfigs);
LOGGER.info("Cluster configs changed: {}", changedConfigs);
- _clusterConfigs = clusterConfigs;
+ _effectiveConfigs = effectiveConfigs;
for (PinotClusterConfigChangeListener listener : _listeners) {
- listener.onChange(changedConfigs, clusterConfigs);
+ listener.onChange(changedConfigs, effectiveConfigs);
}
}
@Override
public synchronized Map<String, String> getClusterConfigs() {
- return _clusterConfigs;
+ return _effectiveConfigs;
}
@Override
public synchronized boolean
registerClusterConfigChangeListener(PinotClusterConfigChangeListener listener) {
LOGGER.info("Registering cluster config change listener: {}",
listener.getClass().getName());
_listeners.add(listener);
// Treat every key as newly added so that the listener picks up the
current values
- listener.onChange(_clusterConfigs.keySet(), _clusterConfigs);
+ listener.onChange(_effectiveConfigs.keySet(), _effectiveConfigs);
return true;
}
+
+ private Map<String, String> getEffectiveConfigs(Map<String, String>
clusterConfigs) {
+ Map<String, String> effectiveConfigs = new
TreeMap<>(CONFIG_KEY_COMPARATOR);
+ effectiveConfigs.putAll(_instanceConfigs);
+ for (Map.Entry<String, String> entry : clusterConfigs.entrySet()) {
+ effectiveConfigs.putIfAbsent(entry.getKey(), entry.getValue());
+ }
+ return Collections.unmodifiableMap(effectiveConfigs);
+ }
+
+ private static Set<String> getChangedEffectiveProperties(Map<String, String>
oldConfigs,
+ Map<String, String> newConfigs) {
+ Set<String> changedConfigs = new TreeSet<>(CONFIG_KEY_COMPARATOR);
+ for (Map.Entry<String, String> entry : newConfigs.entrySet()) {
+ if (!entry.getValue().equals(oldConfigs.get(entry.getKey()))) {
+ changedConfigs.add(entry.getKey());
+ }
+ }
+ for (String key : oldConfigs.keySet()) {
+ if (!newConfigs.containsKey(key)) {
+ changedConfigs.add(key);
+ }
+ }
+ return changedConfigs;
+ }
+
+ private static Map<String, String> snapshot(PinotConfiguration
instanceConfig) {
+ Map<String, String> snapshot = new TreeMap<>(CONFIG_KEY_COMPARATOR);
+ for (String key : instanceConfig.getKeys()) {
+ String value = instanceConfig.getProperty(key);
Review Comment:
[P1] Preserve subset-relative interpolation when replaying instance settings
With the resource-usage accountant enabled, this instance configuration
works before the PR:
```properties
pinot.server.query.accounting.sleep.ms=${sleeptime}
pinot.server.query.accounting.sleeptime=1000
```
`ThreadAccountantUtils.extractAccountingConfig()` strips the accounting
prefix, allowing the initial `QueryMonitorConfig` to resolve `${sleeptime}` to
`1000`. This snapshot reads the property in the full configuration instead,
where that relative reference remains the literal `${sleeptime}`. Registration
then replays it to the update constructor, whose `Integer.parseInt()` throws
`NumberFormatException`. The exception propagates from the accountant listener
registration in `BaseServerStarter:914` (and the equivalent broker path),
preventing startup.
The same issue silently changes enabled OOM protection to disabled when
`pinot.server.query.accounting.oom.enable.killing.query=${enableoom}` and
`pinot.server.query.accounting.enableoom=true`: initialization resolves `true`,
but replay passes the unresolved string to `Boolean.parseBoolean()`, yielding
`false`.
Both numeric and boolean cases were reproduced against exact base/head
sources: the base retains `1000`/`true`; this head throws/changes to `false`.
Please preserve the same interpolation context and typed conversion used during
initialization, and cover both cases with regression tests. Resolving every
property only against the full configuration is insufficient.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
