xiangfu0 commented on code in PR #12039:
URL: https://github.com/apache/pinot/pull/12039#discussion_r4435017612
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java:
##########
@@ -36,56 +39,91 @@ private ServiceStartableUtils() {
private static final Logger LOGGER =
LoggerFactory.getLogger(ServiceStartableUtils.class);
private static final String CLUSTER_CONFIG_ZK_PATH_TEMPLATE =
"/%s/CONFIGS/CLUSTER/%s";
+ private static final String INSTANCE_CONFIG_ZK_PATH_TEMPLATE =
"/%s/CONFIGS/PARTICIPANT/%s";
private static final String PINOT_ALL_CONFIG_KEY_PREFIX = "pinot.all.";
+ private static final String PINOT_TAG_LEVEL_CONFIG_KEY_PREFIX = "pinot.tag.";
private static final String PINOT_INSTANCE_CONFIG_KEY_PREFIX_TEMPLATE =
"pinot.%s.";
+ public static void applyClusterConfig(PinotConfiguration instanceConfig,
String zkAddress, String clusterName,
+ ServiceRole serviceRole) {
+ ZkClient zkClient = getZKClient(instanceConfig, zkAddress);
+ try {
+ applyClusterConfig(instanceConfig, zkClient, clusterName, serviceRole);
+ } finally {
+ zkClient.close();
+ }
+ }
+
/**
* Applies the ZK cluster config to the given instance config if it does not
already exist.
*
* In the ZK cluster config:
* - pinot.all.* will be replaced to role specific config, e.g.
pinot.controller.* for controllers
*/
- public static void applyClusterConfig(PinotConfiguration instanceConfig,
String zkAddress, String clusterName,
+ public static void applyClusterConfig(PinotConfiguration instanceConfig,
ZkClient zkClient, String clusterName,
ServiceRole serviceRole) {
+ ZNRecord clusterConfigZNRecord =
+ zkClient.readData(String.format(CLUSTER_CONFIG_ZK_PATH_TEMPLATE,
clusterName, clusterName), true);
+ if (clusterConfigZNRecord == null) {
+ LOGGER.warn("Failed to find cluster config for cluster: {}, skipping
applying cluster config", clusterName);
+ return;
+ }
+
+ Map<String, String> clusterConfigs =
clusterConfigZNRecord.getSimpleFields();
+ String instanceConfigKeyPrefix =
+ String.format(PINOT_INSTANCE_CONFIG_KEY_PREFIX_TEMPLATE,
serviceRole.name().toLowerCase());
+ for (Map.Entry<String, String> entry : clusterConfigs.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ if (key.startsWith(PINOT_ALL_CONFIG_KEY_PREFIX)) {
+ String instanceConfigKey = instanceConfigKeyPrefix +
key.substring(PINOT_ALL_CONFIG_KEY_PREFIX.length());
+ addConfigIfNotExists(instanceConfig, instanceConfigKey, value);
+ } else {
+ // TODO: Currently it puts all keys to the instance config. Consider
standardizing instance config keys and
+ // only put keys with the instance config key prefix.
+ addConfigIfNotExists(instanceConfig, key, value);
+ }
+ }
+ }
+
+ public static ZkClient getZKClient(PinotConfiguration instanceConfig, String
zkAddress) {
int zkClientSessionConfig =
instanceConfig.getProperty(CommonConstants.Helix.ZkClient.ZK_CLIENT_SESSION_TIMEOUT_MS_CONFIG,
CommonConstants.Helix.ZkClient.DEFAULT_SESSION_TIMEOUT_MS);
int zkClientConnectionTimeoutMs =
instanceConfig.getProperty(CommonConstants.Helix.ZkClient.ZK_CLIENT_CONNECTION_TIMEOUT_MS_CONFIG,
CommonConstants.Helix.ZkClient.DEFAULT_CONNECT_TIMEOUT_MS);
- ZkClient zkClient = new ZkClient.Builder()
- .setZkSerializer(new ZNRecordSerializer())
- .setZkServer(zkAddress)
- .setConnectionTimeout(zkClientConnectionTimeoutMs)
- .setSessionTimeout(zkClientSessionConfig)
- .build();
+ ZkClient zkClient = new ZkClient.Builder().setZkSerializer(new
ZNRecordSerializer()).setZkServer(zkAddress)
+
.setConnectionTimeout(zkClientConnectionTimeoutMs).setSessionTimeout(zkClientSessionConfig).build();
zkClient.waitUntilConnected(zkClientConnectionTimeoutMs,
TimeUnit.MILLISECONDS);
+ return zkClient;
+ }
- try {
- ZNRecord clusterConfigZNRecord =
- zkClient.readData(String.format(CLUSTER_CONFIG_ZK_PATH_TEMPLATE,
clusterName, clusterName), true);
- if (clusterConfigZNRecord == null) {
- LOGGER.warn("Failed to find cluster config for cluster: {}, skipping
applying cluster config", clusterName);
- return;
- }
+ /**
+ * Overrides the instance config with the tenant configs if the tenant is
tagged on the instance.
+ */
+ public static void applyTenantConfigs(String instanceId, ZkClient zkClient,
String clusterName,
+ PinotConfiguration instanceConfig) {
+ ZNRecord instanceConfigZNRecord =
+ zkClient.readData(String.format(INSTANCE_CONFIG_ZK_PATH_TEMPLATE,
clusterName, instanceId), true);
+ if (instanceConfigZNRecord == null) {
+ LOGGER.warn("Failed to find instance config for instance: {}, skipping
overriding tenant configs", instanceId);
+ return;
+ }
+ InstanceConfig instanceZKConfig = new
InstanceConfig(instanceConfigZNRecord);
+ Set<String> instanceTags = instanceZKConfig.getTags().stream()
+ .map(PinotConfiguration::relaxPropertyName)
+ .collect(Collectors.toSet());
- Map<String, String> clusterConfigs =
clusterConfigZNRecord.getSimpleFields();
- String instanceConfigKeyPrefix =
- String.format(PINOT_INSTANCE_CONFIG_KEY_PREFIX_TEMPLATE,
serviceRole.name().toLowerCase());
- for (Map.Entry<String, String> entry : clusterConfigs.entrySet()) {
- String key = entry.getKey();
- String value = entry.getValue();
- if (key.startsWith(PINOT_ALL_CONFIG_KEY_PREFIX)) {
- String instanceConfigKey = instanceConfigKeyPrefix +
key.substring(PINOT_ALL_CONFIG_KEY_PREFIX.length());
- addConfigIfNotExists(instanceConfig, instanceConfigKey, value);
- } else {
- // TODO: Currently it puts all keys to the instance config. Consider
standardizing instance config keys and
- // only put keys with the instance config key prefix.
- addConfigIfNotExists(instanceConfig, key, value);
+ for (String key : instanceConfig.getKeys()) {
+ if (key.startsWith(PINOT_TAG_LEVEL_CONFIG_KEY_PREFIX)) {
+ String instanceConfigKey =
key.substring(PINOT_TAG_LEVEL_CONFIG_KEY_PREFIX.length());
+ String tag = instanceConfigKey.substring(0,
instanceConfigKey.indexOf('.'));
Review Comment:
**Blocking:** a `pinot.tag.` key with no `.` after the tag (e.g.
`pinot.tag.myTenant_OFFLINE`) makes `indexOf('.')` return -1 and throws
`StringIndexOutOfBoundsException`. This happens before the tag-membership
check, so it hits instances that don't carry the tag too. `POST
/cluster/configs` accepts any key, and `applyClusterConfig` copies
non-`pinot.all.*` keys into every instance. So one typo breaks `init()` on
every broker and server at the next restart. Reproduced locally at this SHA.
Please skip and warn on malformed keys, and add a unit test.
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
##########
pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java:
##########
@@ -135,34 +136,44 @@ public void init(PinotConfiguration brokerConf)
// Remove all white-spaces from the list of zkServers (if any).
_zkServers =
brokerConf.getProperty(Helix.CONFIG_OF_ZOOKEEPR_SERVER).replaceAll("\\s+", "");
_clusterName = brokerConf.getProperty(Helix.CONFIG_OF_CLUSTER_NAME);
- ServiceStartableUtils.applyClusterConfig(_brokerConf, _zkServers,
_clusterName, ServiceRole.BROKER);
+ ZkClient zkClient = ServiceStartableUtils.getZKClient(_brokerConf,
_zkServers);
+ try {
+ ServiceStartableUtils.applyClusterConfig(_brokerConf, zkClient,
_clusterName, ServiceRole.BROKER);
+ _hostname = brokerConf.getProperty(Broker.CONFIG_OF_BROKER_HOSTNAME);
+ if (_hostname == null) {
+ _hostname =
+ _brokerConf.getProperty(Helix.SET_INSTANCE_ID_TO_HOSTNAME_KEY,
false) ? NetUtils.getHostnameOrAddress()
+ : NetUtils.getHostAddress();
+ }
+
+ if
(_brokerConf.getProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_PORT,
+ MultiStageQueryRunner.DEFAULT_QUERY_RUNNER_PORT) == 0) {
+
_brokerConf.setProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_PORT,
NetUtils.findOpenPort());
+ }
+ _listenerConfigs = ListenerConfigUtil.buildBrokerConfigs(brokerConf);
+
+ // Override multi-stage query runner hostname if not set explicitly
+ if
(!_brokerConf.containsKey(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_HOSTNAME)) {
+
_brokerConf.setProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_HOSTNAME,
_hostname);
+ }
+ _port = _listenerConfigs.get(0).getPort();
+
+ _instanceId = _brokerConf.getProperty(Broker.CONFIG_OF_BROKER_ID);
+ if (_instanceId == null) {
+ _instanceId = _brokerConf.getProperty(Helix.Instance.INSTANCE_ID_KEY);
+ }
+ if (_instanceId == null) {
+ _instanceId = Helix.PREFIX_OF_BROKER_INSTANCE + _hostname + "_" +
_port;
+ }
- if (_brokerConf.getProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_PORT,
- MultiStageQueryRunner.DEFAULT_QUERY_RUNNER_PORT) == 0) {
- _brokerConf.setProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_PORT,
NetUtils.findOpenPort());
+ ServiceStartableUtils.applyTenantConfigs(_instanceId, zkClient,
_clusterName, _brokerConf);
Review Comment:
The overrides are applied after hostname, port, `_listenerConfigs`, the
query-runner port/hostname and `_instanceId` are already resolved. Tag-level
values for those keys get written into the conf but are silently ignored. The
server builds its admin listeners after the override (`BaseServerStarter`
L187), so the same key behaves differently on brokers and servers. Please
either reject or document the keys that can't be overridden, or apply the
overrides before these values are resolved.
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java:
##########
@@ -36,56 +39,91 @@ private ServiceStartableUtils() {
private static final Logger LOGGER =
LoggerFactory.getLogger(ServiceStartableUtils.class);
private static final String CLUSTER_CONFIG_ZK_PATH_TEMPLATE =
"/%s/CONFIGS/CLUSTER/%s";
+ private static final String INSTANCE_CONFIG_ZK_PATH_TEMPLATE =
"/%s/CONFIGS/PARTICIPANT/%s";
private static final String PINOT_ALL_CONFIG_KEY_PREFIX = "pinot.all.";
+ private static final String PINOT_TAG_LEVEL_CONFIG_KEY_PREFIX = "pinot.tag.";
private static final String PINOT_INSTANCE_CONFIG_KEY_PREFIX_TEMPLATE =
"pinot.%s.";
+ public static void applyClusterConfig(PinotConfiguration instanceConfig,
String zkAddress, String clusterName,
+ ServiceRole serviceRole) {
+ ZkClient zkClient = getZKClient(instanceConfig, zkAddress);
+ try {
+ applyClusterConfig(instanceConfig, zkClient, clusterName, serviceRole);
+ } finally {
+ zkClient.close();
+ }
+ }
+
/**
* Applies the ZK cluster config to the given instance config if it does not
already exist.
*
* In the ZK cluster config:
* - pinot.all.* will be replaced to role specific config, e.g.
pinot.controller.* for controllers
*/
- public static void applyClusterConfig(PinotConfiguration instanceConfig,
String zkAddress, String clusterName,
+ public static void applyClusterConfig(PinotConfiguration instanceConfig,
ZkClient zkClient, String clusterName,
ServiceRole serviceRole) {
+ ZNRecord clusterConfigZNRecord =
+ zkClient.readData(String.format(CLUSTER_CONFIG_ZK_PATH_TEMPLATE,
clusterName, clusterName), true);
+ if (clusterConfigZNRecord == null) {
+ LOGGER.warn("Failed to find cluster config for cluster: {}, skipping
applying cluster config", clusterName);
+ return;
+ }
+
+ Map<String, String> clusterConfigs =
clusterConfigZNRecord.getSimpleFields();
+ String instanceConfigKeyPrefix =
+ String.format(PINOT_INSTANCE_CONFIG_KEY_PREFIX_TEMPLATE,
serviceRole.name().toLowerCase());
+ for (Map.Entry<String, String> entry : clusterConfigs.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ if (key.startsWith(PINOT_ALL_CONFIG_KEY_PREFIX)) {
+ String instanceConfigKey = instanceConfigKeyPrefix +
key.substring(PINOT_ALL_CONFIG_KEY_PREFIX.length());
+ addConfigIfNotExists(instanceConfig, instanceConfigKey, value);
+ } else {
+ // TODO: Currently it puts all keys to the instance config. Consider
standardizing instance config keys and
+ // only put keys with the instance config key prefix.
+ addConfigIfNotExists(instanceConfig, key, value);
+ }
+ }
+ }
+
+ public static ZkClient getZKClient(PinotConfiguration instanceConfig, String
zkAddress) {
int zkClientSessionConfig =
instanceConfig.getProperty(CommonConstants.Helix.ZkClient.ZK_CLIENT_SESSION_TIMEOUT_MS_CONFIG,
CommonConstants.Helix.ZkClient.DEFAULT_SESSION_TIMEOUT_MS);
int zkClientConnectionTimeoutMs =
instanceConfig.getProperty(CommonConstants.Helix.ZkClient.ZK_CLIENT_CONNECTION_TIMEOUT_MS_CONFIG,
CommonConstants.Helix.ZkClient.DEFAULT_CONNECT_TIMEOUT_MS);
- ZkClient zkClient = new ZkClient.Builder()
- .setZkSerializer(new ZNRecordSerializer())
- .setZkServer(zkAddress)
- .setConnectionTimeout(zkClientConnectionTimeoutMs)
- .setSessionTimeout(zkClientSessionConfig)
- .build();
+ ZkClient zkClient = new ZkClient.Builder().setZkSerializer(new
ZNRecordSerializer()).setZkServer(zkAddress)
+
.setConnectionTimeout(zkClientConnectionTimeoutMs).setSessionTimeout(zkClientSessionConfig).build();
zkClient.waitUntilConnected(zkClientConnectionTimeoutMs,
TimeUnit.MILLISECONDS);
+ return zkClient;
+ }
- try {
- ZNRecord clusterConfigZNRecord =
- zkClient.readData(String.format(CLUSTER_CONFIG_ZK_PATH_TEMPLATE,
clusterName, clusterName), true);
- if (clusterConfigZNRecord == null) {
- LOGGER.warn("Failed to find cluster config for cluster: {}, skipping
applying cluster config", clusterName);
- return;
- }
+ /**
+ * Overrides the instance config with the tenant configs if the tenant is
tagged on the instance.
+ */
+ public static void applyTenantConfigs(String instanceId, ZkClient zkClient,
String clusterName,
+ PinotConfiguration instanceConfig) {
+ ZNRecord instanceConfigZNRecord =
+ zkClient.readData(String.format(INSTANCE_CONFIG_ZK_PATH_TEMPLATE,
clusterName, instanceId), true);
+ if (instanceConfigZNRecord == null) {
+ LOGGER.warn("Failed to find instance config for instance: {}, skipping
overriding tenant configs", instanceId);
+ return;
+ }
+ InstanceConfig instanceZKConfig = new
InstanceConfig(instanceConfigZNRecord);
+ Set<String> instanceTags = instanceZKConfig.getTags().stream()
+ .map(PinotConfiguration::relaxPropertyName)
+ .collect(Collectors.toSet());
- Map<String, String> clusterConfigs =
clusterConfigZNRecord.getSimpleFields();
- String instanceConfigKeyPrefix =
- String.format(PINOT_INSTANCE_CONFIG_KEY_PREFIX_TEMPLATE,
serviceRole.name().toLowerCase());
- for (Map.Entry<String, String> entry : clusterConfigs.entrySet()) {
- String key = entry.getKey();
- String value = entry.getValue();
- if (key.startsWith(PINOT_ALL_CONFIG_KEY_PREFIX)) {
- String instanceConfigKey = instanceConfigKeyPrefix +
key.substring(PINOT_ALL_CONFIG_KEY_PREFIX.length());
- addConfigIfNotExists(instanceConfig, instanceConfigKey, value);
- } else {
- // TODO: Currently it puts all keys to the instance config. Consider
standardizing instance config keys and
- // only put keys with the instance config key prefix.
- addConfigIfNotExists(instanceConfig, key, value);
+ for (String key : instanceConfig.getKeys()) {
+ if (key.startsWith(PINOT_TAG_LEVEL_CONFIG_KEY_PREFIX)) {
+ String instanceConfigKey =
key.substring(PINOT_TAG_LEVEL_CONFIG_KEY_PREFIX.length());
+ String tag = instanceConfigKey.substring(0,
instanceConfigKey.indexOf('.'));
+ String tagKey = instanceConfigKey.substring(tag.length() + 1);
+ if (instanceTags.contains(tag)) {
+ instanceConfig.setProperty(tagKey, instanceConfig.getProperty(key));
Review Comment:
`setProperty` overwrites explicit local/CLI values. I verified locally that
`5` from a ZK tag key replaces a local `1000`. That is the opposite of
`applyClusterConfig`'s `addConfigIfNotExists` contract. When an instance
matches several tags with conflicting values, the last key iterated wins.
Please define and document the precedence (local vs. tag vs. cluster, and tag
vs. tag), and make it deterministic.
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
--
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]
