This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new a3e25756e17 Read each broker's instance config once per refresh in
ExternalViewReader (#19567)
a3e25756e17 is described below
commit a3e25756e179e03d97ec346b87f7b585bbd6a3e8
Author: Jinesh Parakh <[email protected]>
AuthorDate: Tue Sep 15 18:14:26 2026 +0530
Read each broker's instance config once per refresh in ExternalViewReader
(#19567)
Signed-off-by: Jinesh Parakh <[email protected]>
---
.../apache/pinot/client/ExternalViewReader.java | 20 +++-
.../pinot/client/ExternalViewReaderTest.java | 103 ++++++++++++++++++++-
2 files changed, 120 insertions(+), 3 deletions(-)
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/ExternalViewReader.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/ExternalViewReader.java
index b5b992eaa91..c1851c04b8c 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/ExternalViewReader.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/ExternalViewReader.java
@@ -78,6 +78,7 @@ public class ExternalViewReader {
public List<String> getLiveBrokers() {
List<String> brokerUrls = new ArrayList<>();
+ Map<String, String> hostPortByBroker = new HashMap<>();
try {
byte[] brokerResourceNodeData =
_zkClient.readData(BROKER_EXTERNAL_VIEW_PATH, true);
brokerResourceNodeData = unpackZnodeIfNecessary(brokerResourceNodeData);
@@ -89,7 +90,7 @@ public class ExternalViewReader {
for (Entry<String, JsonNode> brokerEntry : resource.properties()) {
String brokerName = brokerEntry.getKey();
if (brokerName.startsWith("Broker_") &&
"ONLINE".equals(brokerEntry.getValue().asText())) {
- brokerUrls.add(getHostPort(brokerName));
+ brokerUrls.add(resolveHostPort(brokerName, hostPortByBroker));
}
}
}
@@ -100,6 +101,20 @@ public class ExternalViewReader {
return brokerUrls;
}
+ /// Resolves a broker's address, reading its instance config at most once
per enclosing call.
+ ///
+ /// A broker's address is a property of the broker, not of the table being
examined, but the
+ /// callers walk the broker resource table by table and so meet the same
broker once per table it
+ /// serves. Resolving through this map turns those N x M reads into one per
distinct broker.
+ ///
+ /// The map is supplied by the caller and lives only for that call,
deliberately. An instance
+ /// field would have to be both synchronised, since these methods are called
concurrently, and
+ /// invalidated whenever a broker's host or port changes; a per-call map can
be neither stale nor
+ /// contended, and the redundancy it removes is entirely within a single
traversal anyway.
+ private String resolveHostPort(String brokerName, Map<String, String>
hostPortByBroker) {
+ return hostPortByBroker.computeIfAbsent(brokerName, this::getHostPort);
+ }
+
@VisibleForTesting
String getHostPort(String brokerName) {
// Turn Broker_12.34.56.78_1234 into 12.34.56.78:1234, try InstanceConfig
first, naming convention as backup
@@ -148,6 +163,7 @@ public class ExternalViewReader {
public Map<String, List<String>> getTableToBrokersMap() {
Map<String, Set<String>> brokerUrlsMap = new HashMap<>();
+ Map<String, String> hostPortByBroker = new HashMap<>();
try {
byte[] brokerResourceNodeData =
_zkClient.readData(BROKER_EXTERNAL_VIEW_PATH, true);
brokerResourceNodeData = unpackZnodeIfNecessary(brokerResourceNodeData);
@@ -162,7 +178,7 @@ public class ExternalViewReader {
for (Entry<String, JsonNode> brokerEntry : resource.properties()) {
String brokerName = brokerEntry.getKey();
if (brokerName.startsWith("Broker_") &&
"ONLINE".equals(brokerEntry.getValue().asText())) {
- brokerUrls.add(getHostPort(brokerName));
+ brokerUrls.add(resolveHostPort(brokerName, hostPortByBroker));
}
}
}
diff --git
a/pinot-clients/pinot-java-client/src/test/java/org/apache/pinot/client/ExternalViewReaderTest.java
b/pinot-clients/pinot-java-client/src/test/java/org/apache/pinot/client/ExternalViewReaderTest.java
index 68f50079b7a..ec547bbcab8 100644
---
a/pinot-clients/pinot-java-client/src/test/java/org/apache/pinot/client/ExternalViewReaderTest.java
+++
b/pinot-clients/pinot-java-client/src/test/java/org/apache/pinot/client/ExternalViewReaderTest.java
@@ -26,6 +26,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.io.IOUtils;
import org.mockito.Mock;
@@ -37,7 +38,6 @@ import static org.mockito.Mockito.when;
import static org.mockito.MockitoAnnotations.initMocks;
import static org.testng.Assert.assertEquals;
-
public class ExternalViewReaderTest {
@Mock
@@ -220,4 +220,105 @@ public class ExternalViewReaderTest {
final List<String> brokers = _externalViewReaderUnderTest.getLiveBrokers();
assertEquals(brokers,
Arrays.asList("first.pug-pinot-broker-headless:8090"));
}
+
+ //
-----------------------------------------------------------------------------------------------
+ // Instance-config read amplification
+ //
+ // The broker resource is walked table by table, so the same broker is met
once per table it
+ // serves. Its address is a property of the broker, so every read after the
first returns the same
+ // bytes. The fixtures above cannot show this: they hold one table and one
broker, where N x M and
+ // M are both 1.
+ //
-----------------------------------------------------------------------------------------------
+
+ private static final String BROKER_A = "Broker_10.0.0.1_8099";
+ private static final String BROKER_B = "Broker_10.0.0.2_8099";
+
+ /// Two tables, both served by the same two brokers: four (table, broker)
pairs over two distinct
+ /// brokers.
+ private ExternalViewReader newReaderOverTwoTablesAndTwoBrokers() {
+ String externalView = "{\"mapFields\":{"
+ + "\"table1_OFFLINE\":{\"" + BROKER_A + "\":\"ONLINE\",\"" + BROKER_B
+ "\":\"ONLINE\"},"
+ + "\"table2_OFFLINE\":{\"" + BROKER_A + "\":\"ONLINE\",\"" + BROKER_B
+ "\":\"ONLINE\"}}}";
+ when(_mockZkClient.readData(ExternalViewReader.BROKER_EXTERNAL_VIEW_PATH,
true))
+ .thenReturn("json".getBytes(StandardCharsets.UTF_8));
+ when(_mockZkClient.readData(ExternalViewReader.BROKER_INSTANCE_PATH + "/"
+ BROKER_A, true))
+ .thenReturn(instanceConfig("10.0.0.1",
"8099").getBytes(StandardCharsets.UTF_8));
+ when(_mockZkClient.readData(ExternalViewReader.BROKER_INSTANCE_PATH + "/"
+ BROKER_B, true))
+ .thenReturn(instanceConfig("10.0.0.2",
"8099").getBytes(StandardCharsets.UTF_8));
+ return new ExternalViewReader(_mockZkClient) {
+ @Override
+ protected ByteArrayInputStream getInputStream(byte[]
brokerResourceNodeData) {
+ return new
ByteArrayInputStream(externalView.getBytes(StandardCharsets.UTF_8));
+ }
+ };
+ }
+
+ private static String instanceConfig(String host, String port) {
+ return "{\"id\":\"Broker_" + host + "_" + port + "\",\"simpleFields\":{"
+ + "\"HELIX_HOST\":\"" + host + "\",\"HELIX_PORT\":\"" + port + "\"},"
+ + "\"mapFields\":{},\"listFields\":{}}";
+ }
+
+ @Test
+ public void testGetTableToBrokersMapReadsEachInstanceConfigOncePerCall() {
+ ExternalViewReader reader = newReaderOverTwoTablesAndTwoBrokers();
+
+ reader.getTableToBrokersMap();
+
+ // Two distinct brokers, so two reads -- not the four the four (table,
broker) pairs would imply.
+ Mockito.verify(_mockZkClient, Mockito.times(1))
+ .readData(ExternalViewReader.BROKER_INSTANCE_PATH + "/" + BROKER_A,
true);
+ Mockito.verify(_mockZkClient, Mockito.times(1))
+ .readData(ExternalViewReader.BROKER_INSTANCE_PATH + "/" + BROKER_B,
true);
+ }
+
+ /// Fewer reads, identical result: every table still maps to both broker
addresses.
+ @Test
+ public void testGetLiveBrokersReadsEachInstanceConfigOncePerCall() {
+ ExternalViewReader reader = newReaderOverTwoTablesAndTwoBrokers();
+
+ reader.getLiveBrokers();
+
+ Mockito.verify(_mockZkClient, Mockito.times(1))
+ .readData(ExternalViewReader.BROKER_INSTANCE_PATH + "/" + BROKER_A,
true);
+ Mockito.verify(_mockZkClient, Mockito.times(1))
+ .readData(ExternalViewReader.BROKER_INSTANCE_PATH + "/" + BROKER_B,
true);
+ }
+
+ /// getLiveBrokers returns a List and has always emitted one entry per
(table, broker) pair rather
+ /// than per broker. Reusing a resolved address must not quietly turn that
into a de-duplicated
+ /// list: it is public API, and a caller picking at random would see its
weighting change.
+ @Test
+ public void testGetLiveBrokersStillReturnsOneEntryPerTableBrokerPair() {
+ ExternalViewReader reader = newReaderOverTwoTablesAndTwoBrokers();
+
+ List<String> brokers = reader.getLiveBrokers();
+
+ assertEquals(brokers.size(), 4);
+ assertEquals(Set.copyOf(brokers), Set.of("10.0.0.1:8099",
"10.0.0.2:8099"));
+ }
+
+ @Test
+ public void
testGetTableToBrokersMapResultIsUnchangedByReusingResolvedAddresses() {
+ ExternalViewReader reader = newReaderOverTwoTablesAndTwoBrokers();
+
+ Map<String, List<String>> result = reader.getTableToBrokersMap();
+
+ assertEquals(result.keySet(), Set.of("table1", "table2"));
+ assertEquals(Set.copyOf(result.get("table1")), Set.of("10.0.0.1:8099",
"10.0.0.2:8099"));
+ assertEquals(Set.copyOf(result.get("table2")), Set.of("10.0.0.1:8099",
"10.0.0.2:8099"));
+ }
+
+ /// A fresh call must re-read: the map lives for one call only, so a broker
that moved host or
+ /// port is picked up on the next refresh without any invalidation machinery.
+ @Test
+ public void testResolvedAddressesAreNotRetainedAcrossCalls() {
+ ExternalViewReader reader = newReaderOverTwoTablesAndTwoBrokers();
+
+ reader.getTableToBrokersMap();
+ reader.getTableToBrokersMap();
+
+ Mockito.verify(_mockZkClient, Mockito.times(2))
+ .readData(ExternalViewReader.BROKER_INSTANCE_PATH + "/" + BROKER_A,
true);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
