Jackie-Jiang commented on code in PR #19686:
URL: https://github.com/apache/pinot/pull/19686#discussion_r4174965356
##########
pinot-common/src/main/java/org/apache/pinot/common/config/DefaultClusterConfigChangeHandler.java:
##########
@@ -19,50 +19,112 @@
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 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 = getChangedProperties(_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] Guard disabled routing stats before replaying instance settings
This callback now includes instance-only settings. If the broker
configuration sets
`pinot.broker.adaptive.server.selector.stats.metric.export.interval.ms=10000`
while statistics collection is disabled (the default),
`ServerRoutingStatsManager.init()` returns without creating
`_metricExportFuture`. Registration in `BaseBrokerStarter` then invokes
`onChange()`, whose interval branch calls `_metricExportFuture.cancel(false)`
and throws a `NullPointerException`, aborting broker startup. Before this
change, a file-only export interval did not reach the listener, so this
configuration was harmless.
Please guard the listener when statistics collection is disabled and add a
regression test that registers it with an effective configuration containing an
instance export interval.
--
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]
