Jackie-Jiang commented on code in PR #20058: URL: https://github.com/apache/pinot/pull/20058#discussion_r4393039586
########## pinot-common/src/main/java/org/apache/pinot/common/utils/helix/InstanceCapabilityTracker.java: ########## @@ -0,0 +1,339 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.common.utils.helix; + +import com.google.common.annotations.VisibleForTesting; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.stream.Collectors; +import javax.annotation.Nullable; +import org.apache.helix.HelixAdmin; +import org.apache.helix.HelixManager; +import org.apache.helix.NotificationContext; +import org.apache.helix.api.listeners.BatchMode; +import org.apache.helix.api.listeners.InstanceConfigChangeListener; +import org.apache.helix.api.listeners.PreFetch; +import org.apache.helix.model.InstanceConfig; +import org.apache.helix.zookeeper.datamodel.ZNRecord; +import org.apache.pinot.spi.config.instance.InstanceType; +import org.apache.pinot.spi.utils.CommonConstants; +import org.apache.pinot.spi.utils.InstanceTypeUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Tracks the capabilities that the instances of one type advertise, see [InstanceCapability]. It tracks the +/// capabilities of [InstanceCapability#values()]. +/// +/// The tracker is a Helix [InstanceConfigChangeListener]. Register it with +/// `HelixManager.addInstanceConfigChangeListener`. Each instance advertises its capabilities in the map field +/// `capabilities` of its instance config ([CommonConstants.Helix.Instance#CAPABILITIES_KEY]). For each capability, +/// [#getMinRevision] returns the minimum revision across all the instances of the tracked type in the cluster. A +/// missing map, a missing key or a malformed value counts as revision 0 (not supported). A malformed value is logged +/// as a warning. Instances of other types are ignored. +/// +/// The minimum covers every instance config of the tracked type in the cluster, including the instances that are +/// disabled, not live, or serve other tenants. Such an instance can lower the minimum although it never receives the +/// query. This is fail-safe: a lower minimum only keeps the plan that every version can run. +/// +/// Rollback guard: the map also holds the reserved entry `pinotVersion` +/// ([CommonConstants.Helix.Instance#CAPABILITIES_PINOT_VERSION_KEY]), the Pinot version that wrote the capabilities. +/// The tracker reads the capabilities of an instance only when that entry is string-equal to the simple field +/// `pinotVersion` ([CommonConstants.Helix.Instance#PINOT_VERSION_KEY]) of the same instance config. Otherwise it treats +/// the instance as supporting no capability (revision 0). Every Pinot version rewrites the simple field at startup, +/// but a version that predates the capabilities does not clear the map. So after a rollback to such a version the two +/// values differ, and the stale capabilities are ignored. This is an equality check between two fields of one instance +/// config. It does not order versions and it does not compare the versions of different instances. +/// Residual gap: `PinotHelixResourceManager.updateInstance` reads the instance config and writes the full record back +/// without a version check. When an instance restarts on another version between that read and that write, the write +/// restores the previous simple field `pinotVersion` and the previous map, so stale capabilities can be trusted. The +/// simple field `pinotVersion` has the same problem today. A follow-up can make the controller keep the fields that +/// the instance owns. +/// +/// [#getMinRevision] returns 0 before the first notification and while the cluster has no instance of the tracked +/// type. Helix sends the `INIT` notification when the listener is registered. +/// +/// When the instance config of one instance cannot be read on a `CALLBACK` notification (for example because of a +/// transient ZooKeeper error), the instance counts as revision 0 until its instance config changes again or a new +/// session sends `INIT`. This is fail-safe. +/// +/// When the instances below the revision that this version supports ([InstanceCapability#getRevision()]) change, the +/// tracker logs their ids at INFO, at most [#MAX_LOGGED_INSTANCES] per list. It lists separately the instances that +/// advertise no or an older revision and the instances whose capabilities are ignored because the version stamp does +/// not match. It never logs on the query path. +/// +/// Thread safety: Helix callback threads call [#onInstanceConfigChange] and query threads call [#getMinRevision]. The +/// notification handling is `synchronized` and owns the per-instance bookkeeping. It publishes an immutable snapshot +/// of the minimum revisions through a `volatile` field, so readers never block and see the result of the last +/// processed notification. +@BatchMode(enabled = false) +@PreFetch(enabled = false) +public class InstanceCapabilityTracker implements InstanceConfigChangeListener { + private static final Logger LOGGER = LoggerFactory.getLogger(InstanceCapabilityTracker.class); + /// The maximum number of instance ids that the tracker logs for each capability. + static final int MAX_LOGGED_INSTANCES = 20; + + private final HelixManager _helixManager; + private final String _clusterName; + private final InstanceType _instanceType; + private final List<InstanceCapability> _capabilities; + // Guarded by this + private final Map<String, Map<InstanceCapability, Integer>> _revisionsByInstance = new HashMap<>(); + // Guarded by this. The instances whose capabilities are ignored because the version stamp does not match. + private final Set<String> _staleInstances = new HashSet<>(); + // Guarded by this. For each capability, the sorted ids of the instances below InstanceCapability.getRevision(), + // and the sorted ids of the stale instances, as last logged. + private Map<InstanceCapability, List<String>> _laggingInstances = Map.of(); + private List<String> _loggedStaleInstances = List.of(); + + private HelixAdmin _helixAdmin; + private volatile Map<InstanceCapability, Integer> _minRevisions = Map.of(); + + /// Tracks the instances of `instanceType`. The Helix manager does not need to be connected yet. It must be connected + /// before the tracker is registered. + public InstanceCapabilityTracker(HelixManager helixManager, InstanceType instanceType) { + this(helixManager, instanceType, InstanceCapability.values()); + } + + /// Tracks the given capabilities instead of [InstanceCapability#values()]. + @VisibleForTesting + InstanceCapabilityTracker(HelixManager helixManager, InstanceType instanceType, + Collection<InstanceCapability> capabilities) { + _helixManager = helixManager; + _clusterName = helixManager.getClusterName(); + _instanceType = instanceType; + _capabilities = List.copyOf(capabilities); + } + + /// The Helix manager of the tracked cluster. + public HelixManager getHelixManager() { + return _helixManager; + } + + /// The name of the tracked cluster. + public String getClusterName() { + return _clusterName; + } + + /// The minimum revision of the capability across all the instances of the tracked type, 0 when one of them does not + /// support it or when the cluster has no such instance. Never blocks. + public int getMinRevision(InstanceCapability capability) { + return _minRevisions.getOrDefault(capability, 0); + } + + @Override + public synchronized void onInstanceConfigChange(List<InstanceConfig> instanceConfigs, NotificationContext context) { + if (_helixAdmin == null) { + _helixAdmin = _helixManager.getClusterManagmentTool(); + } + NotificationContext.Type type = context.getType(); + if (type != NotificationContext.Type.INIT && type != NotificationContext.Type.CALLBACK) { + LOGGER.warn("Ignoring notification type: {} for instance config change", type); + return; + } + if (type == NotificationContext.Type.INIT || context.getIsChildChange()) { + // Instances may have been added or removed: read them all. + _revisionsByInstance.clear(); + _staleInstances.clear(); + for (String instance : _helixAdmin.getInstancesInCluster(_clusterName)) { Review Comment: [P2] Keep the tracker fail-safe after a failed full refresh The bookkeeping is cleared before `getInstancesInCluster()`, which can throw. For example, start with server A at revision 1 and B at revision 0: a failed membership read leaves the map empty, and a later config callback for A populates only A, so `computeMinRevisions()` incorrectly publishes 1 while B is still present. Helix catches the callback exception without replaying it. Build replacement state separately and keep the published minimum at 0 after a failed refresh until a complete refresh succeeds; incremental callbacks must not restore a positive result from incomplete membership. Add a regression case for a failed membership read followed by a single-instance callback. ########## pinot-common/src/main/java/org/apache/pinot/common/utils/helix/MultiClusterInstanceCapabilityTracker.java: ########## @@ -0,0 +1,126 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.common.utils.helix; + +import com.google.common.annotations.VisibleForTesting; +import java.util.Collection; +import java.util.List; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CopyOnWriteArraySet; +import org.apache.helix.HelixManager; +import org.apache.helix.PropertyKey; +import org.apache.pinot.spi.config.instance.InstanceType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Tracks the capabilities of the instances of one type across several Helix clusters, for example the local cluster +/// of a broker and the remote clusters that it federates. It holds one [InstanceCapabilityTracker] for each cluster +/// that it watches. +/// +/// [#getMinRevision] returns the minimum revision across all the clusters. A cluster that the caller adds with +/// [#addUnavailableCluster] (for example a remote cluster that failed to connect), or whose tracker failed to +/// register, counts as revision 0. This is fail-safe: a lower minimum only keeps the plan that every version can run. +/// Without any cluster, the minimum is 0 too. +/// +/// An unavailable cluster stays at revision 0 for the life of this instance: nothing retries the connection or the +/// registration. For a broker, a remote cluster that fails to connect at startup pins the minimum to 0 until the +/// broker restarts. Both cases are logged at WARN or ERROR with the cluster name. +/// +/// Lifecycle: add the clusters with [#addCluster] and [#addUnavailableCluster] while the Helix managers are +/// connected, then share the instance. [#close] removes the Helix listeners. Call it before the Helix managers +/// disconnect. +/// +/// Thread safety: the clusters are kept in copy-on-write collections, so query threads may call [#getMinRevision] +/// at any time, also while clusters are added. [#getMinRevision] never blocks. +public class MultiClusterInstanceCapabilityTracker { + private static final Logger LOGGER = LoggerFactory.getLogger(MultiClusterInstanceCapabilityTracker.class); + + private final InstanceType _instanceType; + private final List<InstanceCapability> _capabilities; + private final List<InstanceCapabilityTracker> _trackers = new CopyOnWriteArrayList<>(); + private final Set<String> _unavailableClusters = new CopyOnWriteArraySet<>(); + + /// Tracks the capabilities of [InstanceCapability#values()] for the instances of `instanceType`. + public MultiClusterInstanceCapabilityTracker(InstanceType instanceType) { + this(instanceType, InstanceCapability.values()); + } + + /// Tracks the given capabilities instead of [InstanceCapability#values()]. + @VisibleForTesting + MultiClusterInstanceCapabilityTracker(InstanceType instanceType, Collection<InstanceCapability> capabilities) { + _instanceType = instanceType; + _capabilities = List.copyOf(capabilities); + } + + /// Watches the cluster of the given connected Helix manager: creates an [InstanceCapabilityTracker] and registers it + /// as an instance config change listener. When the registration fails, the cluster counts as revision 0. + public void addCluster(HelixManager helixManager) { + addCluster(new InstanceCapabilityTracker(helixManager, _instanceType, _capabilities)); + } + + @VisibleForTesting + void addCluster(InstanceCapabilityTracker tracker) { + String clusterName = tracker.getClusterName(); + try { + tracker.getHelixManager().addInstanceConfigChangeListener(tracker); + _trackers.add(tracker); + LOGGER.info("Tracking the capabilities of the {} instances of cluster: {}", _instanceType, clusterName); + } catch (Exception e) { + LOGGER.error("Failed to register the InstanceCapabilityTracker for cluster: {}, treating the cluster as " + + "supporting no capability until restart", clusterName, e); + _unavailableClusters.add(clusterName); + } + } + + /// Adds a cluster that cannot be watched, for example a remote cluster that failed to connect. It counts as + /// revision 0 for every capability. + public void addUnavailableCluster(String clusterName) { + LOGGER.warn("Cluster: {} is unavailable, treating it as supporting no capability until restart", clusterName); + _unavailableClusters.add(clusterName); + } + + /// The minimum revision of the capability across the instances of all the clusters. 0 when a cluster is + /// unavailable, when an instance does not support the capability, or when no cluster is tracked. Never blocks. + public int getMinRevision(InstanceCapability capability) { + if (!_unavailableClusters.isEmpty() || _trackers.isEmpty()) { + return 0; + } + int minRevision = Integer.MAX_VALUE; + for (InstanceCapabilityTracker tracker : _trackers) { Review Comment: [P2] Use one tracker snapshot for the empty check and iteration A reader can pass `_trackers.isEmpty()`, then `close()` clears the list before the enhanced-for loop obtains its iterator. The loop visits no trackers and returns `Integer.MAX_VALUE`, falsely reporting support for every revision. The class permits query reads at any time, and the existing close test establishes that an empty/closed tracker should return 0. Capture one snapshot for both the emptiness check and iteration, or explicitly return 0 when no trackers were visited. The current sequential close test does not cover this interleaving. -- 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]
