| /* |
| * 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.solr.cloud; |
| |
| import java.lang.invoke.MethodHandles; |
| import java.util.Arrays; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Locale; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.atomic.AtomicLong; |
| |
| import org.apache.commons.math3.primes.Primes; |
| import org.apache.lucene.util.LuceneTestCase.Slow; |
| import org.apache.solr.client.solrj.SolrClient; |
| import org.apache.solr.client.solrj.impl.HttpSolrClient; |
| import org.apache.solr.client.solrj.request.UpdateRequest; |
| import org.apache.solr.client.solrj.response.QueryResponse; |
| import org.apache.solr.client.solrj.response.UpdateResponse; |
| import org.apache.solr.common.SolrDocument; |
| import org.apache.solr.common.SolrInputDocument; |
| import org.apache.solr.common.cloud.ClusterState; |
| import org.apache.solr.common.cloud.Replica; |
| import org.apache.solr.common.cloud.Slice; |
| import org.apache.solr.common.cloud.ZkStateReader; |
| import org.apache.solr.common.params.ModifiableSolrParams; |
| import org.apache.solr.common.util.NamedList; |
| import org.apache.zookeeper.KeeperException; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| @Slow |
| public class TestStressInPlaceUpdates extends AbstractFullDistribZkTestBase { |
| private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); |
| |
| @BeforeClass |
| public static void beforeSuperClass() throws Exception { |
| schemaString = "schema-inplace-updates.xml"; |
| configString = "solrconfig-tlog.xml"; |
| |
| // sanity check that autocommits are disabled |
| initCore(configString, schemaString); |
| assertEquals(-1, h.getCore().getSolrConfig().getUpdateHandlerInfo().autoCommmitMaxTime); |
| assertEquals(-1, h.getCore().getSolrConfig().getUpdateHandlerInfo().autoSoftCommmitMaxTime); |
| assertEquals(-1, h.getCore().getSolrConfig().getUpdateHandlerInfo().autoCommmitMaxDocs); |
| assertEquals(-1, h.getCore().getSolrConfig().getUpdateHandlerInfo().autoSoftCommmitMaxDocs); |
| } |
| |
| public TestStressInPlaceUpdates() { |
| super(); |
| sliceCount = 1; |
| fixShardCount(3); |
| } |
| |
| protected final ConcurrentHashMap<Integer, DocInfo> model = new ConcurrentHashMap<>(); |
| protected Map<Integer, DocInfo> committedModel = new HashMap<>(); |
| protected long snapshotCount; |
| protected long committedModelClock; |
| protected int clientIndexUsedForCommit; |
| protected volatile int lastId; |
| protected final String field = "val_l"; |
| |
| private void initModel(int ndocs) { |
| for (int i = 0; i < ndocs; i++) { |
| // seed versions w/-1 so "from scratch" adds/updates will fail optimistic concurrency checks |
| // if some other thread beats us to adding the id |
| model.put(i, new DocInfo(-1L, 0, 0)); |
| } |
| committedModel.putAll(model); |
| } |
| |
| SolrClient leaderClient = null; |
| |
| @Test |
| @ShardsFixed(num = 3) |
| // commented out on: 17-Feb-2019 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // 09-Apr-2018 |
| public void stressTest() throws Exception { |
| waitForRecoveriesToFinish(true); |
| |
| this.leaderClient = getClientForLeader(); |
| assertNotNull("Couldn't obtain client for the leader of the shard", this.leaderClient); |
| |
| final int commitPercent = 5 + random().nextInt(20); |
| final int softCommitPercent = 30 + random().nextInt(75); // what percent of the commits are soft |
| final int deletePercent = 4 + random().nextInt(25); |
| final int deleteByQueryPercent = random().nextInt(8); |
| final int ndocs = atLeast(5); |
| int nWriteThreads = 5 + random().nextInt(12); |
| int fullUpdatePercent = 5 + random().nextInt(50); |
| |
| // query variables |
| final int percentRealtimeQuery = 75; |
| // number of cumulative read/write operations by all threads |
| final AtomicLong operations = new AtomicLong(5000); |
| int nReadThreads = 5 + random().nextInt(12); |
| |
| |
| /** // testing |
| final int commitPercent = 5; |
| final int softCommitPercent = 100; // what percent of the commits are soft |
| final int deletePercent = 0; |
| final int deleteByQueryPercent = 50; |
| final int ndocs = 10; |
| int nWriteThreads = 10; |
| |
| final int maxConcurrentCommits = nWriteThreads; // number of committers at a time... it should be <= maxWarmingSearchers |
| |
| // query variables |
| final int percentRealtimeQuery = 101; |
| final AtomicLong operations = new AtomicLong(50000); // number of query operations to perform in total |
| int nReadThreads = 10; |
| |
| int fullUpdatePercent = 20; |
| **/ |
| |
| if (log.isInfoEnabled()) { |
| log.info("{}", Arrays.asList |
| ("commitPercent", commitPercent, "softCommitPercent", softCommitPercent, |
| "deletePercent", deletePercent, "deleteByQueryPercent", deleteByQueryPercent, |
| "ndocs", ndocs, "nWriteThreads", nWriteThreads, "percentRealtimeQuery", percentRealtimeQuery, |
| "operations", operations, "nReadThreads", nReadThreads)); |
| } |
| |
| initModel(ndocs); |
| |
| List<Thread> threads = new ArrayList<>(); |
| |
| for (int i = 0; i < nWriteThreads; i++) { |
| Thread thread = new Thread("WRITER" + i) { |
| Random rand = new Random(random().nextInt()); |
| |
| @Override |
| public void run() { |
| try { |
| while (operations.decrementAndGet() > 0) { |
| int oper = rand.nextInt(50); |
| |
| if (oper < commitPercent) { |
| Map<Integer, DocInfo> newCommittedModel; |
| long version; |
| |
| synchronized (TestStressInPlaceUpdates.this) { |
| // take a snapshot of the model |
| // this is safe to do w/o synchronizing on the model because it's a ConcurrentHashMap |
| newCommittedModel = new HashMap<>(model); |
| version = snapshotCount++; |
| |
| int chosenClientIndex = rand.nextInt(clients.size()); |
| |
| if (rand.nextInt(100) < softCommitPercent) { |
| log.info("softCommit start"); |
| clients.get(chosenClientIndex).commit(true, true, true); |
| log.info("softCommit end"); |
| } else { |
| log.info("hardCommit start"); |
| clients.get(chosenClientIndex).commit(); |
| log.info("hardCommit end"); |
| } |
| |
| // install this model snapshot only if it's newer than the current one |
| if (version >= committedModelClock) { |
| if (VERBOSE) { |
| log.info("installing new committedModel version={}", committedModelClock); |
| } |
| clientIndexUsedForCommit = chosenClientIndex; |
| committedModel = newCommittedModel; |
| committedModelClock = version; |
| } |
| } |
| continue; |
| } |
| |
| int id; |
| |
| if (rand.nextBoolean()) { |
| id = rand.nextInt(ndocs); |
| } else { |
| id = lastId; // reuse the last ID half of the time to force more race conditions |
| } |
| |
| // set the lastId before we actually change it sometimes to try and |
| // uncover more race conditions between writing and reading |
| boolean before = rand.nextBoolean(); |
| if (before) { |
| lastId = id; |
| } |
| |
| DocInfo info = model.get(id); |
| |
| // yield after getting the next version to increase the odds of updates happening out of order |
| if (rand.nextBoolean()) Thread.yield(); |
| |
| if (oper < commitPercent + deletePercent + deleteByQueryPercent) { |
| final boolean dbq = (oper >= commitPercent + deletePercent); |
| final String delType = dbq ? "DBI": "DBQ"; |
| log.info("{} id {}: {}", delType, id, info); |
| |
| Long returnedVersion = null; |
| |
| try { |
| returnedVersion = deleteDocAndGetVersion(Integer.toString(id), params("_version_", Long.toString(info.version)), dbq); |
| log.info("{}: Deleting id={}, version={}. Returned version={}" |
| , delType, id, info.version, returnedVersion); |
| } catch (RuntimeException e) { |
| if (e.getMessage() != null && e.getMessage().contains("version conflict") |
| || e.getMessage() != null && e.getMessage().contains("Conflict")) { |
| // Its okay for a leader to reject a concurrent request |
| log.warn("Conflict during {}, rejected id={}, {}", delType, id, e); |
| returnedVersion = null; |
| } else { |
| throw e; |
| } |
| } |
| |
| // only update model if update had no conflict & the version is newer |
| synchronized (model) { |
| DocInfo currInfo = model.get(id); |
| if (null != returnedVersion && |
| (Math.abs(returnedVersion.longValue()) > Math.abs(currInfo.version))) { |
| model.put(id, new DocInfo(returnedVersion.longValue(), 0, 0)); |
| } |
| } |
| |
| } else { |
| int val1 = info.intFieldValue; |
| long val2 = info.longFieldValue; |
| int nextVal1 = val1; |
| long nextVal2 = val2; |
| |
| int addOper = rand.nextInt(30); |
| Long returnedVersion; |
| if (addOper < fullUpdatePercent || info.version <= 0) { // if document was never indexed or was deleted |
| // FULL UPDATE |
| nextVal1 = Primes.nextPrime(val1 + 1); |
| nextVal2 = nextVal1 * 1000000000l; |
| try { |
| returnedVersion = addDocAndGetVersion("id", id, "title_s", "title" + id, "val1_i_dvo", nextVal1, "val2_l_dvo", nextVal2, "_version_", info.version); |
| log.info("FULL: Writing id={}, val=[{},{}], version={}, Prev was=[{},{}]. Returned version={}" |
| ,id, nextVal1, nextVal2, info.version, val1, val2, returnedVersion); |
| |
| } catch (RuntimeException e) { |
| if (e.getMessage() != null && e.getMessage().contains("version conflict") |
| || e.getMessage() != null && e.getMessage().contains("Conflict")) { |
| // Its okay for a leader to reject a concurrent request |
| log.warn("Conflict during full update, rejected id={}, {}", id, e); |
| returnedVersion = null; |
| } else { |
| throw e; |
| } |
| } |
| } else { |
| // PARTIAL |
| nextVal2 = val2 + val1; |
| try { |
| returnedVersion = addDocAndGetVersion("id", id, "val2_l_dvo", map("inc", String.valueOf(val1)), "_version_", info.version); |
| log.info("PARTIAL: Writing id={}, val=[{},{}], version={}, Prev was=[{},{}]. Returned version={}" |
| ,id, nextVal1, nextVal2, info.version, val1, val2, returnedVersion); |
| } catch (RuntimeException e) { |
| if (e.getMessage() != null && e.getMessage().contains("version conflict") |
| || e.getMessage() != null && e.getMessage().contains("Conflict")) { |
| // Its okay for a leader to reject a concurrent request |
| log.warn("Conflict during partial update, rejected id={}, {}", id, e); |
| } else if (e.getMessage() != null && e.getMessage().contains("Document not found for update.") |
| && e.getMessage().contains("id="+id)) { |
| log.warn("Attempted a partial update for a recently deleted document, rejected id={}, {}", id, e); |
| } else { |
| throw e; |
| } |
| returnedVersion = null; |
| } |
| } |
| |
| // only update model if update had no conflict & the version is newer |
| synchronized (model) { |
| DocInfo currInfo = model.get(id); |
| if (null != returnedVersion && |
| (Math.abs(returnedVersion.longValue()) > Math.abs(currInfo.version))) { |
| model.put(id, new DocInfo(returnedVersion.longValue(), nextVal1, nextVal2)); |
| } |
| |
| } |
| } |
| |
| if (!before) { |
| lastId = id; |
| } |
| } |
| } catch (Throwable e) { |
| operations.set(-1L); |
| log.error("", e); |
| throw new RuntimeException(e); |
| } |
| } |
| }; |
| |
| threads.add(thread); |
| |
| } |
| |
| // Read threads |
| for (int i = 0; i < nReadThreads; i++) { |
| Thread thread = new Thread("READER" + i) { |
| Random rand = new Random(random().nextInt()); |
| |
| @SuppressWarnings("unchecked") |
| @Override |
| public void run() { |
| try { |
| while (operations.decrementAndGet() >= 0) { |
| // bias toward a recently changed doc |
| int id = rand.nextInt(100) < 25 ? lastId : rand.nextInt(ndocs); |
| |
| // when indexing, we update the index, then the model |
| // so when querying, we should first check the model, and then the index |
| |
| boolean realTime = rand.nextInt(100) < percentRealtimeQuery; |
| DocInfo expected; |
| |
| if (realTime) { |
| expected = model.get(id); |
| } else { |
| synchronized (TestStressInPlaceUpdates.this) { |
| expected = committedModel.get(id); |
| } |
| } |
| |
| if (VERBOSE) { |
| log.info("querying id {}", id); |
| } |
| ModifiableSolrParams params = new ModifiableSolrParams(); |
| if (realTime) { |
| params.set("wt", "json"); |
| params.set("qt", "/get"); |
| params.set("ids", Integer.toString(id)); |
| } else { |
| params.set("wt", "json"); |
| params.set("q", "id:" + Integer.toString(id)); |
| params.set("omitHeader", "true"); |
| } |
| |
| int clientId = rand.nextInt(clients.size()); |
| if (!realTime) clientId = clientIndexUsedForCommit; |
| |
| QueryResponse response = clients.get(clientId).query(params); |
| if (response.getResults().size() == 0) { |
| // there's no info we can get back with a delete, so not much we can check without further synchronization |
| } else if (response.getResults().size() == 1) { |
| final SolrDocument actual = response.getResults().get(0); |
| final String msg = "Realtime=" + realTime + ", expected=" + expected + ", actual=" + actual; |
| assertNotNull(msg, actual); |
| |
| final Long foundVersion = (Long) actual.getFieldValue("_version_"); |
| assertNotNull(msg, foundVersion); |
| assertTrue(msg + "... solr doc has non-positive version???", |
| 0 < foundVersion.longValue()); |
| final Integer intVal = (Integer) actual.getFieldValue("val1_i_dvo"); |
| assertNotNull(msg, intVal); |
| |
| final Long longVal = (Long) actual.getFieldValue("val2_l_dvo"); |
| assertNotNull(msg, longVal); |
| |
| assertTrue(msg + " ...solr returned older version then model. " + |
| "should not be possible given the order of operations in writer threads", |
| Math.abs(expected.version) <= foundVersion.longValue()); |
| |
| if (foundVersion.longValue() == expected.version) { |
| assertEquals(msg, expected.intFieldValue, intVal.intValue()); |
| assertEquals(msg, expected.longFieldValue, longVal.longValue()); |
| } |
| |
| // Some things we can assert about any Doc returned from solr, |
| // even if it's newer then our (expected) model information... |
| |
| assertTrue(msg + " ...how did a doc in solr get a non positive intVal?", |
| 0 < intVal); |
| assertTrue(msg + " ...how did a doc in solr get a non positive longVal?", |
| 0 < longVal); |
| assertEquals(msg + " ...intVal and longVal in solr doc are internally (modulo) inconsistent w/eachother", |
| 0, (longVal % intVal)); |
| |
| // NOTE: when foundVersion is greater then the version read from the model, |
| // it's not possible to make any assertions about the field values in solr relative to the |
| // field values in the model -- ie: we can *NOT* assert expected.longFieldVal <= doc.longVal |
| // |
| // it's tempting to think that this would be possible if we changed our model to preserve the |
| // "old" valuess when doing a delete, but that's still no garuntee because of how oportunistic |
| // concurrency works with negative versions: When adding a doc, we can assert that it must not |
| // exist with version<0, but we can't assert that the *reason* it doesn't exist was because of |
| // a delete with the specific version of "-42". |
| // So a wrtier thread might (1) prep to add a doc for the first time with "intValue=1,_version_=-1", |
| // and that add may succeed and (2) return some version X which is put in the model. but |
| // inbetween #1 and #2 other threads may have added & deleted the doc repeatedly, updating |
| // the model with intValue=7,_version_=-42, and a reader thread might meanwhile read from the |
| // model before #2 and expect intValue=5, but get intValue=1 from solr (with a greater version) |
| |
| } else { |
| fail(String.format(Locale.ENGLISH, "There were more than one result: {}", response)); |
| } |
| } |
| } catch (Throwable e) { |
| operations.set(-1L); |
| log.error("", e); |
| throw new RuntimeException(e); |
| } |
| } |
| }; |
| |
| threads.add(thread); |
| } |
| // Start all threads |
| for (Thread thread : threads) { |
| thread.start(); |
| } |
| |
| for (Thread thread : threads) { |
| thread.join(); |
| } |
| |
| { // final pass over uncommitted model with RTG |
| |
| for (SolrClient client : clients) { |
| for (Map.Entry<Integer,DocInfo> entry : model.entrySet()) { |
| final Integer id = entry.getKey(); |
| final DocInfo expected = entry.getValue(); |
| final SolrDocument actual = client.getById(id.toString()); |
| |
| String msg = "RTG: " + id + "=" + expected; |
| if (null == actual) { |
| // a deleted or non-existent document |
| // sanity check of the model agrees... |
| assertTrue(msg + " is deleted/non-existent in Solr, but model has non-neg version", |
| expected.version < 0); |
| assertEquals(msg + " is deleted/non-existent in Solr", expected.intFieldValue, 0); |
| assertEquals(msg + " is deleted/non-existent in Solr", expected.longFieldValue, 0); |
| } else { |
| msg = msg + " <==VS==> " + actual; |
| assertEquals(msg, expected.intFieldValue, actual.getFieldValue("val1_i_dvo")); |
| assertEquals(msg, expected.longFieldValue, actual.getFieldValue("val2_l_dvo")); |
| assertEquals(msg, expected.version, actual.getFieldValue("_version_")); |
| assertTrue(msg + " doc exists in solr, but version is negative???", |
| 0 < expected.version); |
| } |
| } |
| } |
| } |
| |
| { // do a final search and compare every result with the model |
| |
| // because commits don't provide any sort of concrete versioning (or optimistic concurrency constraints) |
| // there's no way to garuntee that our committedModel matches what was in Solr at the time of the last commit. |
| // It's possible other threads made additional writes to solr before the commit was processed, but after |
| // the committedModel variable was assigned it's new value. |
| // |
| // what we can do however, is commit all completed updates, and *then* compare solr search results |
| // against the (new) committed model.... |
| |
| waitForThingsToLevelOut(30); // NOTE: this does an automatic commit for us & ensures replicas are up to date |
| committedModel = new HashMap<>(model); |
| |
| // first, prune the model of any docs that have negative versions |
| // ie: were never actually added, or were ultimately deleted. |
| for (int i = 0; i < ndocs; i++) { |
| DocInfo info = committedModel.get(i); |
| if (info.version < 0) { |
| // first, a quick sanity check of the model itself... |
| assertEquals("Inconsistent int value in model for deleted doc" + i + "=" + info, |
| 0, info.intFieldValue); |
| assertEquals("Inconsistent long value in model for deleted doc" + i + "=" + info, |
| 0L, info.longFieldValue); |
| |
| committedModel.remove(i); |
| } |
| } |
| |
| for (SolrClient client : clients) { |
| QueryResponse rsp = client.query(params("q","*:*", "sort", "id asc", "rows", ndocs+"")); |
| for (SolrDocument actual : rsp.getResults()) { |
| final Integer id = Integer.parseInt(actual.getFieldValue("id").toString()); |
| final DocInfo expected = committedModel.get(id); |
| |
| assertNotNull("Doc found but missing/deleted from model: " + actual, expected); |
| |
| final String msg = "Search: " + id + "=" + expected + " <==VS==> " + actual; |
| assertEquals(msg, expected.intFieldValue, actual.getFieldValue("val1_i_dvo")); |
| assertEquals(msg, expected.longFieldValue, actual.getFieldValue("val2_l_dvo")); |
| assertEquals(msg, expected.version, actual.getFieldValue("_version_")); |
| assertTrue(msg + " doc exists in solr, but version is negative???", |
| 0 < expected.version); |
| |
| // also sanity check the model (which we already know matches the doc) |
| assertEquals("Inconsistent (modulo) values in model for id " + id + "=" + expected, |
| 0, (expected.longFieldValue % expected.intFieldValue)); |
| } |
| assertEquals(committedModel.size(), rsp.getResults().getNumFound()); |
| } |
| } |
| } |
| |
| /** |
| * Used for storing the info for a document in an in-memory model. |
| */ |
| private static class DocInfo { |
| long version; |
| int intFieldValue; |
| long longFieldValue; |
| |
| public DocInfo(long version, int val1, long val2) { |
| assert version != 0; // must either be real positive version, or negative deleted version/indicator |
| this.version = version; |
| this.intFieldValue = val1; |
| this.longFieldValue = val2; |
| } |
| |
| @Override |
| public String toString() { |
| return "[version=" + version + ", intValue=" + intFieldValue + ",longValue=" + longFieldValue + "]"; |
| } |
| } |
| |
| @SuppressWarnings("rawtypes") |
| protected long addDocAndGetVersion(Object... fields) throws Exception { |
| SolrInputDocument doc = new SolrInputDocument(); |
| addFields(doc, fields); |
| |
| ModifiableSolrParams params = new ModifiableSolrParams(); |
| params.add("versions", "true"); |
| |
| UpdateRequest ureq = new UpdateRequest(); |
| ureq.setParams(params); |
| ureq.add(doc); |
| UpdateResponse resp; |
| |
| // send updates to leader, to avoid SOLR-8733 |
| resp = ureq.process(leaderClient); |
| |
| long returnedVersion = Long.parseLong(((NamedList) resp.getResponse().get("adds")).getVal(0).toString()); |
| assertTrue("Due to SOLR-8733, sometimes returned version is 0. Let us assert that we have successfully" |
| + " worked around that problem here.", returnedVersion > 0); |
| return returnedVersion; |
| } |
| |
| @SuppressWarnings("rawtypes") |
| protected long deleteDocAndGetVersion(String id, ModifiableSolrParams params, boolean deleteByQuery) throws Exception { |
| params.add("versions", "true"); |
| |
| UpdateRequest ureq = new UpdateRequest(); |
| ureq.setParams(params); |
| if (deleteByQuery) { |
| ureq.deleteByQuery("id:"+id); |
| } else { |
| ureq.deleteById(id); |
| } |
| UpdateResponse resp; |
| // send updates to leader, to avoid SOLR-8733 |
| resp = ureq.process(leaderClient); |
| |
| String key = deleteByQuery? "deleteByQuery": "deletes"; |
| long returnedVersion = Long.parseLong(((NamedList) resp.getResponse().get(key)).getVal(0).toString()); |
| assertTrue("Due to SOLR-8733, sometimes returned version is 0. Let us assert that we have successfully" |
| + " worked around that problem here.", returnedVersion < 0); |
| return returnedVersion; |
| } |
| |
| /** |
| * Method gets the SolrClient for the leader replica. This is needed for a workaround for SOLR-8733. |
| */ |
| public SolrClient getClientForLeader() throws KeeperException, InterruptedException { |
| ZkStateReader zkStateReader = cloudClient.getZkStateReader(); |
| cloudClient.getZkStateReader().forceUpdateCollection(DEFAULT_COLLECTION); |
| ClusterState clusterState = cloudClient.getZkStateReader().getClusterState(); |
| Replica leader = null; |
| Slice shard1 = clusterState.getCollection(DEFAULT_COLLECTION).getSlice(SHARD1); |
| leader = shard1.getLeader(); |
| |
| for (int i = 0; i < clients.size(); i++) { |
| String leaderBaseUrl = zkStateReader.getBaseUrlForNodeName(leader.getNodeName()); |
| if (((HttpSolrClient) clients.get(i)).getBaseURL().startsWith(leaderBaseUrl)) |
| return clients.get(i); |
| } |
| |
| return null; |
| } |
| } |
