You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by jl...@apache.org on 2023/11/03 22:54:12 UTC
(pinot) branch fix-npe-for-minimizeDataMovement created (now d9832a4ac4)
This is an automated email from the ASF dual-hosted git repository.
jlli pushed a change to branch fix-npe-for-minimizeDataMovement
in repository https://gitbox.apache.org/repos/asf/pinot.git
at d9832a4ac4 Fix the NPE in minimizeDataMovement instance assignment strategy
This branch includes the following new commits:
new d9832a4ac4 Fix the NPE in minimizeDataMovement instance assignment strategy
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
(pinot) 01/01: Fix the NPE in minimizeDataMovement instance assignment strategy
Posted by jl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jlli pushed a commit to branch fix-npe-for-minimizeDataMovement
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit d9832a4ac443f7cd4fed1f3fa4db55f49507a86a
Author: jlli_LinkedIn <jl...@linkedin.com>
AuthorDate: Fri Nov 3 15:48:31 2023 -0700
Fix the NPE in minimizeDataMovement instance assignment strategy
---
.../InstanceReplicaGroupPartitionSelector.java | 11 ++-
.../InstanceReplicaGroupPartitionSelectorTest.java | 105 +++++++++++++++++++++
2 files changed, 111 insertions(+), 5 deletions(-)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java
index b6c62ac12e..de1e681d17 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java
@@ -72,11 +72,17 @@ public class InstanceReplicaGroupPartitionSelector extends InstancePartitionSele
Preconditions.checkState(numReplicaGroups > 0, "Number of replica-groups must be positive");
Map<Integer, List<Integer>> poolToReplicaGroupIdsMap = new TreeMap<>();
Map<Integer, Integer> replicaGroupIdToPoolMap = new TreeMap<>();
+ Map<Integer, Set<String>> poolToCandidateInstancesMap = new TreeMap<>();
for (int replicaId = 0; replicaId < numReplicaGroups; replicaId++) {
// Pick one pool for each replica-group based on the table name hash
int pool = pools.get((tableNameHash + replicaId) % numPools);
poolToReplicaGroupIdsMap.computeIfAbsent(pool, k -> new ArrayList<>()).add(replicaId);
replicaGroupIdToPoolMap.put(replicaId, pool);
+
+ Set<String> candidateInstances =
+ poolToCandidateInstancesMap.computeIfAbsent(pool, k -> new LinkedHashSet<>());
+ List<InstanceConfig> instanceConfigsInPool = poolToInstanceConfigsMap.get(pool);
+ instanceConfigsInPool.forEach(k -> candidateInstances.add(k.getInstanceName()));
}
LOGGER.info("Selecting {} replica-groups from pool: {} for table: {}", numReplicaGroups, poolToReplicaGroupIdsMap,
_tableNameWithType);
@@ -132,7 +138,6 @@ public class InstanceReplicaGroupPartitionSelector extends InstancePartitionSele
int existingNumReplicaGroups = _existingInstancePartitions.getNumReplicaGroups();
int numCommonReplicaGroups = Math.min(numReplicaGroups, existingNumReplicaGroups);
- Map<Integer, Set<String>> poolToCandidateInstancesMap = new TreeMap<>();
Map<Integer, Set<String>> replicaGroupIdToExistingInstancesMap = new TreeMap<>();
// Step 1: find out the replica groups and their existing instances,
// so that these instances can be filtered out and won't be chosen for the other replica group.
@@ -142,10 +147,6 @@ public class InstanceReplicaGroupPartitionSelector extends InstancePartitionSele
// Skip the replica group if it's no longer needed.
continue;
}
- Set<String> candidateInstances =
- poolToCandidateInstancesMap.computeIfAbsent(pool, k -> new LinkedHashSet<>());
- List<InstanceConfig> instanceConfigsInPool = poolToInstanceConfigsMap.get(pool);
- instanceConfigsInPool.forEach(k -> candidateInstances.add(k.getInstanceName()));
for (int partitionId = 0; partitionId < existingNumPartitions; partitionId++) {
List<String> existingInstances = _existingInstancePartitions.getInstances(partitionId, replicaGroupId);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelectorTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelectorTest.java
new file mode 100644
index 0000000000..4f982c45e6
--- /dev/null
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelectorTest.java
@@ -0,0 +1,105 @@
+/**
+ * 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.controller.helix.core.assignment.instance;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.text.StringSubstitutor;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.pinot.common.assignment.InstancePartitions;
+import org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
+import org.testng.annotations.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+public class InstanceReplicaGroupPartitionSelectorTest {
+
+ private static final String instanceConfigTemplate =
+ "{\n" + " \"id\": \"Server_pinot-server-${serverName}.pinot-server-headless.pinot.svc.cluster.local_8098\",\n"
+ + " \"simpleFields\": {\n" + " \"HELIX_ENABLED\": \"true\",\n"
+ + " \"HELIX_ENABLED_TIMESTAMP\": \"1688959934305\",\n"
+ + " \"HELIX_HOST\": \"pinot-server-${serverName}.pinot-server-headless.pinot.svc.cluster.local\",\n"
+ + " \"HELIX_PORT\": \"8098\",\n" + " \"adminPort\": \"8097\",\n" + " \"grpcPort\": \"8090\",\n"
+ + " \"queryMailboxPort\": \"46347\",\n" + " \"queryServerPort\": \"45031\",\n"
+ + " \"shutdownInProgress\": \"false\"\n" + " },\n" + " \"mapFields\": {\n"
+ + " \"SYSTEM_RESOURCE_INFO\": {\n" + " \"numCores\": \"16\",\n"
+ + " \"totalMemoryMB\": \"126976\",\n" + " \"maxHeapSizeMB\": \"65536\"\n" + " },\n"
+ + " \"pool\": {\n" + " \"DefaultTenant_OFFLINE\": \"${pool}\",\n"
+ + " \"${poolName}\": \"${pool}\",\n" + " \"AllReplicationGroups\": \"1\"\n" + " }\n" + " },\n"
+ + " \"listFields\": {\n" + " \"TAG_LIST\": [\n" + " \"DefaultTenant_OFFLINE\",\n"
+ + " \"DefaultTenant_REALTIME\",\n" + " \"${poolName}\",\n" + " \"AllReplicationGroups\"\n"
+ + " ]\n" + " }\n" + "}";
+
+ @Test
+ public void testSelectInstances() throws JsonProcessingException {
+ ObjectMapper objectMapper = new ObjectMapper();
+ String existingPartitionsJson =
+ " {\n" + " \"instancePartitionsName\": \"0f97dac8-4123-47c6-9a4d-b8ce039c5ea5_OFFLINE\",\n"
+ + " \"partitionToInstancesMap\": {\n" + " \"0_0\": [\n"
+ + " \"Server_pinot-server-rg0-0.pinot-server-headless.pinot.svc.cluster.local_8098\",\n"
+ + " \"Server_pinot-server-rg0-1.pinot-server-headless.pinot.svc.cluster.local_8098\"\n"
+ + " ]\n" + " }\n" + " }\n";
+ InstancePartitions existing = objectMapper.readValue(existingPartitionsJson, InstancePartitions.class);
+ InstanceReplicaGroupPartitionConfig config =
+ new InstanceReplicaGroupPartitionConfig(true, 0, 2, 2, 1, 2, true, null);
+
+ InstanceReplicaGroupPartitionSelector selector =
+ new InstanceReplicaGroupPartitionSelector(config, "tableNameBlah", existing);
+
+ String[] serverNames = {"rg0-0", "rg0-1", "rg1-0", "rg1-1"};
+ String[] poolNumbers = {"0", "0", "1", "1"};
+ String[] poolNames = {"FirstHalfReplicationGroups", "FirstHalfReplicationGroups", "SecondHalfReplicationGroups",
+ "SecondHalfReplicationGroups"};
+ Map<Integer, List<InstanceConfig>> poolToInstanceConfigsMap = new HashMap<>();
+
+ for (int i = 0; i < serverNames.length; i++) {
+ Map<String, String> valuesMap = new HashMap<>();
+ valuesMap.put("serverName", serverNames[i]);
+ valuesMap.put("pool", poolNumbers[i]);
+ valuesMap.put("poolName", poolNames[i]);
+
+ StringSubstitutor substitutor = new StringSubstitutor(valuesMap);
+ String resolvedString = substitutor.replace(instanceConfigTemplate);
+
+ ZNRecord znRecord = objectMapper.readValue(resolvedString, ZNRecord.class);
+ int poolNumber = Integer.parseInt(poolNumbers[i]);
+ poolToInstanceConfigsMap.computeIfAbsent(poolNumber, k -> new ArrayList<>()).add(new InstanceConfig(znRecord));
+ }
+ InstancePartitions assignedPartitions = new InstancePartitions("0f97dac8-4123-47c6-9a4d-b8ce039c5ea5_OFFLINE");
+ selector.selectInstances(poolToInstanceConfigsMap, assignedPartitions);
+
+ String expectedInstancePartitions =
+ " {\n" + " \"instancePartitionsName\": \"0f97dac8-4123-47c6-9a4d-b8ce039c5ea5_OFFLINE\",\n"
+ + " \"partitionToInstancesMap\": {\n" + " \"0_0\": [\n"
+ + " \"Server_pinot-server-rg0-0.pinot-server-headless.pinot.svc.cluster.local_8098\",\n"
+ + " \"Server_pinot-server-rg0-1.pinot-server-headless.pinot.svc.cluster.local_8098\"\n"
+ + " ],\n" + " \"0_1\": [\n"
+ + " \"Server_pinot-server-rg1-0.pinot-server-headless.pinot.svc.cluster.local_8098\",\n"
+ + " \"Server_pinot-server-rg1-1.pinot-server-headless.pinot.svc.cluster.local_8098\"\n"
+ + " ]\n" + " }\n" + " }\n";
+ InstancePartitions expectedPartitions =
+ objectMapper.readValue(expectedInstancePartitions, InstancePartitions.class);
+ assert assignedPartitions.equals(expectedPartitions);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org