Jackie-Jiang commented on code in PR #20058: URL: https://github.com/apache/pinot/pull/20058#discussion_r4485791902
########## pinot-common/src/main/java/org/apache/pinot/common/utils/helix/InstanceCapability.java: ########## @@ -0,0 +1,150 @@ +/** + * 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.base.Preconditions; +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 org.apache.pinot.spi.config.instance.InstanceType; + + +/// A feature that a Pinot instance advertises in its instance config, with a revision. +/// +/// Any component (server, broker, controller, minion) may advertise capabilities. Each capability names the area of +/// the feature in its own name (for example `MSE_` for the multi-stage engine) and declares the instance type that +/// advertises it. At startup an instance writes the map field `capabilities` +/// (`CommonConstants.Helix.Instance.CAPABILITIES_KEY`) of its instance config with +/// `HelixHelper.updateCapabilities`: the key is the name of a capability of its instance type and the value is its +/// [#revision()] as a decimal integer, for example `{"MSE_FEATURE": "1"}`. Another component reads the map, +/// for example with [InstanceCapabilityTracker], to decide whether it can use a feature that instances of an older +/// version do not support. A missing map, a missing key or a malformed value means revision 0: not supported. +/// +/// The broker reads the capabilities of the servers from the instance configs that its routing manager already reads, +/// see `RoutingManager.getMinServerCapabilityRevision`. Other components can compute the same minimum over their own +/// set of instances with [InstanceCapabilityTracker]. +/// +/// The map also holds the reserved entry `pinotVersion`, the Pinot version that wrote it. A reader ignores the map +/// when that entry is not string-equal to the simple field `pinotVersion` of the same instance config, so the +/// capabilities of a version do not survive a rollback to a version that predates them. See +/// [InstanceCapabilityTracker]. +/// +/// The capabilities of this version are the constants of this class, listed in [#values()]. Contract: +/// - The revision of a capability only grows. Revision `n` implies support for everything that revisions below `n` +/// added. A change that old instances with the same capability cannot handle increments the revision. +/// - A revision is not a Pinot version. Nothing compares Pinot version strings. +/// - A new feature that old instances cannot handle adds a constant with revision 1, and adds it to [#values()] in +/// the same change as the code that supports the feature. An instance must never advertise a feature it cannot run. +/// - The names are persisted in ZooKeeper: never rename or remove a capability. Review Comment: Optional documentation cleanup: allow capability retirement after the compatibility window This still says a capability must never be removed, while the updated PR description correctly distinguishes permanent name identity from advertisement lifetime. Please document that consumers/fallback branches and eventually advertisements can be retired once no supported reader or rollback target requires the key. A completed rollout alone is not sufficient if an older supported reader still checks it, because an absent key means revision 0. Never rename a persisted capability or reuse its name for another feature. Full-map replacement at startup already supports removing retired advertisements. ########## pinot-common/src/main/java/org/apache/pinot/common/utils/helix/HelixHelper.java: ########## @@ -719,6 +720,33 @@ public static boolean updatePinotVersion(InstanceConfig instanceConfig) { return false; } + /// Writes the capabilities that an instance of the given type advertises in this version + /// ([InstanceCapability#values()]) to the map field `capabilities` of the instance config, together with the + /// reserved entry `pinotVersion` ([CommonConstants.Helix.Instance#CAPABILITIES_PINOT_VERSION_KEY]) set to the Pinot + /// version of this build. Readers ignore the capabilities when that entry differs from the simple field + /// `pinotVersion`, which [#updatePinotVersion] writes. Any instance type can call it. Returns whether the field + /// changed. + public static boolean updateCapabilities(InstanceConfig instanceConfig, InstanceType instanceType) { + return updateCapabilities(instanceConfig, instanceType, InstanceCapability.values()); + } + + /// Same as [#updateCapabilities(InstanceConfig, InstanceType)] with the given capabilities instead of + /// [InstanceCapability#values()]. An extension passes the capabilities of Pinot together with its own, see + /// [InstanceCapability]. Fails when two capabilities have the same name. + public static boolean updateCapabilities(InstanceConfig instanceConfig, InstanceType instanceType, + Collection<InstanceCapability> capabilities) { + ZNRecord record = instanceConfig.getRecord(); + InstanceCapability.checkUniqueNames(List.copyOf(capabilities)); + Map<String, String> currentCapabilities = InstanceCapability.toMapField(capabilities, instanceType); + currentCapabilities.put(CommonConstants.Helix.Instance.CAPABILITIES_PINOT_VERSION_KEY, PinotVersion.VERSION); Review Comment: Optional cleanup: make capability publication self-contained Could updateCapabilities() also call updatePinotVersion() and include that change in its returned flag? The current startup callers are correct, but this would remove their required two-call sequence and ensure extension callers produce a reader-valid record with one call. Also consolidate duplicate-name validation and map conversion into one internal path: this currently copies the collection for checkUniqueNames(), then iterates the original collection again for toMapField(). Keep duplicate detection across all instance types before filtering, and preserve full-map replacement so a downgrade or removal of an advertised key replaces the old value rather than retaining it. ########## pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java: ########## @@ -445,6 +453,17 @@ private void processInstanceConfigChangeInternal() { Helix.ZkClient.RETRY_INTERVAL_MS); long fetchInstanceConfigsEndTimeMs = System.currentTimeMillis(); + // Feed the capability tracker with the same snapshot, so that it reads no ZooKeeper data of its own. A failure here + // must not stop the routing update: the tracker keeps its previous result. Review Comment: [P2] Invalidate capability support after a failed refresh Keeping the previous routing snapshot does not guarantee that the same server IDs still run the same binaries. For example, after the tracker reports revision 1, a server can restart under the same ID on an older build; if the subsequent getChildren() fails after retries, the cached positive minimum survives even though that server no longer supports the feature. ClusterChangeMediator catches the failure without immediately replaying the event, so the result can persist until a later refresh. Please invalidate capability support to 0 on a failed config fetch or tracker update, and restore it only after a complete successful snapshot. Reset the comparison bookkeeping as well: clearing only _minRevisions would let the unchanged-snapshot early return prevent recovery. A regression test should cover a failed refresh followed by successful recovery. The registry is currently empty, but this matters for the new API's fail-safe guarantee. ########## pinot-common/src/main/java/org/apache/pinot/common/utils/helix/InstanceCapabilityTracker.java: ########## @@ -0,0 +1,252 @@ +/** + * 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.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; +import javax.annotation.Nullable; +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; + + +/// Computes, for each capability, the minimum revision across a set of instances of one type, see +/// [InstanceCapability]. +/// +/// The tracker does not read ZooKeeper and does not listen to Helix. Its owner calls [#update] with the instance +/// config records that it already reads, for example the broker routing manager on each instance config change. Each +/// call replaces the whole set: the tracker keeps the records of the tracked type and ignores the others. The owner +/// decides which instances count. A wider set is fail-safe: an instance that never receives the work can only lower +/// the minimum, and a lower minimum only keeps the plan that every version can run. +/// +/// Each instance advertises its capabilities in the map field `capabilities` of its instance config +/// ([CommonConstants.Helix.Instance#CAPABILITIES_KEY]). The tracker reads every entry of the map, so it also tracks +/// the capabilities that an extension defines (see [InstanceCapability]). A missing map, a missing key or a malformed +/// value counts as revision 0 (not supported). [#getMinRevision] returns 0 before the first [#update] and while the +/// set has no instance of the tracked type. +/// +/// 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 cannot detect a rollback between two builds with the same version +/// string. 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. +/// +/// Logging: when the parsed capabilities of the set change, the tracker logs at INFO the instances below the revision +/// that this version supports, for the capabilities given to the constructor, at most [#MAX_LOGGED_INSTANCES] per +/// list. It lists separately the instances whose capabilities are ignored because the version stamp does not match. +/// It logs a malformed value at WARN when the capabilities of that instance change. A call that changes nothing logs +/// nothing. +/// +/// Thread safety: [#update] is `synchronized`. It publishes an immutable snapshot of the minimum revisions through a +/// `volatile` field, so [#getMinRevision] never blocks and sees the result of the last [#update]. +public class InstanceCapabilityTracker { + private static final Logger LOGGER = LoggerFactory.getLogger(InstanceCapabilityTracker.class); + /// The maximum number of instance ids that the tracker logs for each list. + static final int MAX_LOGGED_INSTANCES = 20; + /// The revision stored for a malformed value. Counts as 0. + private static final int MALFORMED_REVISION = -1; + + private final String _clusterName; + private final InstanceType _instanceType; + private final List<InstanceCapability> _loggedCapabilities; + // Guarded by this. For each tracked instance, the revision of each capability that it advertises. + private Map<String, Map<String, Integer>> _revisionsByInstance = Map.of(); + // Guarded by this. The instances whose capabilities are ignored because the version stamp does not match. + private Set<String> _staleInstances = Set.of(); + + private volatile Map<String, Integer> _minRevisions = Map.of(); + + /// Tracks the instances of `instanceType` in the given cluster. Logs the lagging instances for the capabilities of + /// [InstanceCapability#values()]. + public InstanceCapabilityTracker(String clusterName, InstanceType instanceType) { + this(clusterName, instanceType, InstanceCapability.values()); + } + + /// Same as [#InstanceCapabilityTracker(String, InstanceType)], but logs the lagging instances for the given + /// capabilities, for example the capabilities of Pinot and those of an extension. Every capability that the + /// instances advertise is tracked either way. + public InstanceCapabilityTracker(String clusterName, InstanceType instanceType, + Collection<InstanceCapability> loggedCapabilities) { + _clusterName = clusterName; + _instanceType = instanceType; + List<InstanceCapability> capabilities = new ArrayList<>(); + for (InstanceCapability capability : loggedCapabilities) { + if (capability.instanceType() == instanceType) { + capabilities.add(capability); + } + } + _loggedCapabilities = List.copyOf(capabilities); + } + + /// The minimum revision of the capability across the tracked instances, 0 when one of them does not support it or + /// when there is no tracked instance. Never blocks. + public int getMinRevision(InstanceCapability capability) { + return _minRevisions.getOrDefault(capability.name(), 0); + } + + /// Replaces the set of instances with the given instance config records. Keeps the records of the tracked type and + /// ignores the others. Never throws because of the content of a record. + public synchronized void update(Collection<ZNRecord> instanceConfigRecords) { + Map<String, Map<String, Integer>> revisionsByInstance = new HashMap<>(); + Set<String> staleInstances = new TreeSet<>(); + for (ZNRecord record : instanceConfigRecords) { + String instanceId = record.getId(); + if (InstanceTypeUtils.getInstanceType(instanceId) != _instanceType) { + continue; + } + if (hasStaleCapabilities(record)) { + staleInstances.add(instanceId); + } + Map<String, Integer> revisions = readRevisions(record); + revisionsByInstance.put(instanceId, revisions); + if (revisions.containsValue(MALFORMED_REVISION) && !revisions.equals(_revisionsByInstance.get(instanceId))) { + LOGGER.warn("Malformed capability revisions for instance: {} of cluster: {}, treating them as 0: {}", + instanceId, _clusterName, record.getMapField(CommonConstants.Helix.Instance.CAPABILITIES_KEY)); + } + } + if (revisionsByInstance.equals(_revisionsByInstance) && staleInstances.equals(_staleInstances)) { + return; + } + _revisionsByInstance = revisionsByInstance; + _staleInstances = staleInstances; + Map<String, Integer> minRevisions = computeMinRevisions(revisionsByInstance); + if (!minRevisions.equals(_minRevisions)) { + LOGGER.info("Minimum capability revisions across {} {} instances of cluster: {} changed from {} to {}", + revisionsByInstance.size(), _instanceType, _clusterName, _minRevisions, minRevisions); + _minRevisions = minRevisions; + } + logLaggingInstances(); + } + + /// The sorted ids of the tracked instances whose capabilities are ignored because the version stamp does not match. + @VisibleForTesting + synchronized List<String> getStaleInstances() { + return List.copyOf(_staleInstances); + } + + /// For each capability that some instance advertises, the minimum revision across all the instances. A capability + /// with minimum 0 has no entry. + private static Map<String, Integer> computeMinRevisions(Map<String, Map<String, Integer>> revisionsByInstance) { + if (revisionsByInstance.isEmpty()) { + return Map.of(); + } + Map<String, Integer> minRevisions = new HashMap<>(); + // Every instance must advertise a capability, so the candidates are the capabilities of any one instance. + for (String name : revisionsByInstance.values().iterator().next().keySet()) { + int minRevision = Integer.MAX_VALUE; + for (Map<String, Integer> revisions : revisionsByInstance.values()) { + minRevision = Math.min(minRevision, revisions.getOrDefault(name, 0)); + } + if (minRevision > 0) { + minRevisions.put(name, minRevision); + } + } + return Map.copyOf(minRevisions); + } + + /// Logs at INFO, for each logged capability, the instances below the revision that this version supports, in two + /// lists: the instances that advertise no or an older revision, and the instances whose capabilities are ignored + /// because the version stamp does not match. + private void logLaggingInstances() { + Map<String, Map<String, Integer>> sortedRevisions = new TreeMap<>(_revisionsByInstance); Review Comment: Optional performance cleanup: skip diagnostic work that cannot produce output Return before building this TreeMap when _loggedCapabilities is empty or INFO is disabled. In particular, the current empty OSS registry still sorts the server map on a changed snapshot even though the following loop has no iterations. For nonempty registries, the diagnostic lists collect every lagging instance and only truncate when formatting; counters plus at most MAX_LOGGED_INSTANCES examples per category would retain the same counts and deterministic examples with less temporary storage. Separately, computeMinRevisions() can stop scanning a candidate once its minimum reaches <= 0. These are refresh-path improvements; query-time lookups are already inexpensive. -- 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]
