xiangfu0 commented on code in PR #20058:
URL: https://github.com/apache/pinot/pull/20058#discussion_r4381923763
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java:
##########
@@ -530,6 +530,8 @@ private void updateInstanceConfigIfNeeded(ServerConf
serverConf) {
serverConf.getMultiStageMailboxPort());
}
updated |= HelixHelper.updatePinotVersion(instanceConfig);
+ updated |= HelixHelper.updateCapabilities(instanceConfig,
+ org.apache.pinot.spi.config.instance.InstanceType.SERVER);
Review Comment:
Same minor nit as in HelixHelper: this inline FQN could be an import for
readability per the repo convention. Purely stylistic.
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/helix/InstanceCapability.java:
##########
@@ -0,0 +1,123 @@
+/**
+ * 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.List;
+import java.util.Map;
+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
+/// [#getRevision()] 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 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.
+///
+/// Two capabilities are equal when their names are equal. Instances are
immutable and thread-safe.
+public final class InstanceCapability {
+ private static final List<InstanceCapability> VALUES = List.of();
+
+ private final String _name;
+ private final InstanceType _instanceType;
+ private final int _revision;
+
+ /// Package-private: production capabilities are the constants of this
class. Tests create their own.
+ InstanceCapability(String name, InstanceType instanceType, int revision) {
+ // Upper case keeps the name apart from the reserved entry pinotVersion of
the capabilities map.
+ Preconditions.checkArgument(name.matches("[A-Z][A-Z0-9_]*"), "Capability
name must be upper case: %s", name);
+ Preconditions.checkArgument(revision > 0, "Revision of capability %s must
be positive: %s", name, revision);
+ _name = name;
+ _instanceType = instanceType;
+ _revision = revision;
+ }
+
+ /// Every capability of this version.
+ public static List<InstanceCapability> values() {
Review Comment:
Optional: since equality is by name and the contract is
never-rename/never-duplicate, an assertion that `VALUES` has no two
capabilities with the same name would catch a future constant-definition
mistake early. Not needed while the registry is empty, but cheap insurance when
capabilities start landing.
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/helix/MultiClusterInstanceCapabilityTracker.java:
##########
@@ -0,0 +1,125 @@
+/**
+ * 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()) {
Review Comment:
Minor: this guard reads `_unavailableClusters` and `_trackers` separately,
so if `addUnavailableCluster` runs concurrently with a `getMinRevision` call
during startup, the loop can briefly return a higher-than-0 revision before
converging to 0 on the next call. It's self-correcting and the class documents
add-then-share as single-threaded setup, so this is fine as-is — just flagging
it. If you want strict safety during concurrent startup, re-checking
`_unavailableClusters.isEmpty()` after the loop would close it.
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/helix/HelixHelper.java:
##########
@@ -719,6 +721,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,
+ org.apache.pinot.spi.config.instance.InstanceType instanceType) {
Review Comment:
Minor/nit: CLAUDE.md prefers an import over an inline fully-qualified name.
Since `InstanceType` isn't imported here yet, consider adding `import
org.apache.pinot.spi.config.instance.InstanceType;` and using the short name in
both overloads (and the `[#updateCapabilities(...)]` javadoc reference). Not
blocking.
--
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]
