| /* |
| * 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.io.IOException; |
| import java.lang.invoke.MethodHandles; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Random; |
| import java.util.Set; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.stream.Collectors; |
| |
| import org.apache.lucene.util.LuceneTestCase.Slow; |
| import org.apache.solr.JSONTestUtil; |
| import org.apache.solr.SolrTestCaseJ4.SuppressSSL; |
| import org.apache.solr.client.solrj.SolrClient; |
| import org.apache.solr.client.solrj.SolrQuery; |
| import org.apache.solr.client.solrj.SolrServerException; |
| import org.apache.solr.client.solrj.impl.HttpSolrClient; |
| import org.apache.solr.client.solrj.request.CollectionAdminRequest; |
| import org.apache.solr.client.solrj.request.CoreAdminRequest; |
| import org.apache.solr.client.solrj.request.QueryRequest; |
| import org.apache.solr.client.solrj.request.UpdateRequest; |
| import org.apache.solr.client.solrj.response.CoreAdminResponse; |
| import org.apache.solr.client.solrj.response.QueryResponse; |
| import org.apache.solr.common.SolrDocument; |
| import org.apache.solr.common.SolrDocumentList; |
| import org.apache.solr.common.SolrInputDocument; |
| import org.apache.solr.common.cloud.DocCollection; |
| import org.apache.solr.common.cloud.Replica; |
| import org.apache.solr.common.cloud.ZkCoreNodeProps; |
| import org.apache.solr.common.cloud.ZkStateReader; |
| import org.apache.solr.common.params.ModifiableSolrParams; |
| import org.apache.solr.common.util.NamedList; |
| import org.apache.solr.update.processor.DistributedUpdateProcessor; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import static org.apache.solr.update.processor.DistributedUpdateProcessor.DISTRIB_FROM; |
| import static org.apache.solr.update.processor.DistributingUpdateProcessorFactory.DISTRIB_UPDATE_PARAM; |
| |
| @Slow |
| @SuppressSSL(bugUrl = "https://issues.apache.org/jira/browse/SOLR-5776") |
| public class DistributedVersionInfoTest extends SolrCloudTestCase { |
| |
| private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); |
| |
| @BeforeClass |
| public static void setupCluster() throws Exception { |
| configureCluster(3) |
| .addConfig("conf", configset("cloud-minimal")) |
| .configure(); |
| } |
| |
| private static final String COLLECTION = "c8n_vers_1x3"; |
| |
| @Test |
| public void testReplicaVersionHandling() throws Exception { |
| |
| final String shardId = "shard1"; |
| |
| CollectionAdminRequest.createCollection(COLLECTION, "conf", 1, 3) |
| .processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT); |
| |
| final ZkStateReader stateReader = cluster.getSolrClient().getZkStateReader(); |
| stateReader.waitForState(COLLECTION, DEFAULT_TIMEOUT, TimeUnit.SECONDS, |
| (n, c) -> DocCollection.isFullyActive(n, c, 1, 3)); |
| |
| final Replica leader = stateReader.getLeaderRetry(COLLECTION, shardId); |
| |
| // start by reloading the empty collection so we try to calculate the max from an empty index |
| reloadCollection(leader, COLLECTION); |
| |
| sendDoc(1); |
| cluster.getSolrClient().commit(COLLECTION); |
| |
| // verify doc is on the leader and replica |
| final List<Replica> notLeaders = stateReader.getClusterState().getCollection(COLLECTION).getReplicas() |
| .stream() |
| .filter(r -> r.getCoreName().equals(leader.getCoreName()) == false) |
| .collect(Collectors.toList()); |
| assertDocsExistInAllReplicas(leader, notLeaders, COLLECTION, 1, 1, null); |
| |
| // get max version from the leader and replica |
| Replica replica = notLeaders.get(0); |
| Long maxOnLeader = getMaxVersionFromIndex(leader); |
| Long maxOnReplica = getMaxVersionFromIndex(replica); |
| assertEquals("leader and replica should have same max version: " + maxOnLeader, maxOnLeader, maxOnReplica); |
| |
| // send the same doc but with a lower version than the max in the index |
| try (SolrClient client = getHttpSolrClient(replica.getCoreUrl())) { |
| String docId = String.valueOf(1); |
| SolrInputDocument doc = new SolrInputDocument(); |
| doc.setField("id", docId); |
| doc.setField("_version_", maxOnReplica - 1); // bad version!!! |
| |
| // simulate what the leader does when sending a doc to a replica |
| ModifiableSolrParams params = new ModifiableSolrParams(); |
| params.set(DISTRIB_UPDATE_PARAM, DistributedUpdateProcessor.DistribPhase.FROMLEADER.toString()); |
| params.set(DISTRIB_FROM, leader.getCoreUrl()); |
| |
| UpdateRequest req = new UpdateRequest(); |
| req.setParams(params); |
| req.add(doc); |
| |
| log.info("Sending doc with out-of-date version ({}) document directly to replica", maxOnReplica -1); |
| |
| client.request(req); |
| client.commit(); |
| |
| Long docVersion = getVersionFromIndex(replica, docId); |
| assertEquals("older version should have been thrown away", maxOnReplica, docVersion); |
| } |
| |
| reloadCollection(leader, COLLECTION); |
| |
| maxOnLeader = getMaxVersionFromIndex(leader); |
| maxOnReplica = getMaxVersionFromIndex(replica); |
| assertEquals("leader and replica should have same max version after reload", maxOnLeader, maxOnReplica); |
| |
| // now start sending docs while collection is reloading |
| |
| delQ("*:*"); |
| commit(); |
| |
| final Set<Integer> deletedDocs = new HashSet<>(); |
| final AtomicInteger docsSent = new AtomicInteger(0); |
| final Random rand = new Random(5150); |
| Thread docSenderThread = new Thread() { |
| public void run() { |
| |
| // brief delay before sending docs |
| try { |
| Thread.sleep(rand.nextInt(30)+1); |
| } catch (InterruptedException e) {} |
| |
| for (int i=0; i < 1000; i++) { |
| if (i % (rand.nextInt(20)+1) == 0) { |
| try { |
| Thread.sleep(rand.nextInt(50)+1); |
| } catch (InterruptedException e) {} |
| } |
| |
| int docId = i+1; |
| try { |
| sendDoc(docId); |
| docsSent.incrementAndGet(); |
| } catch (Exception e) {} |
| } |
| } |
| }; |
| |
| Thread reloaderThread = new Thread() { |
| public void run() { |
| try { |
| Thread.sleep(rand.nextInt(300)+1); |
| } catch (InterruptedException e) {} |
| |
| for (int i=0; i < 3; i++) { |
| try { |
| reloadCollection(leader, COLLECTION); |
| } catch (Exception e) {} |
| |
| try { |
| Thread.sleep(rand.nextInt(300)+300); |
| } catch (InterruptedException e) {} |
| } |
| } |
| }; |
| |
| Thread deleteThread = new Thread() { |
| public void run() { |
| |
| // brief delay before sending docs |
| try { |
| Thread.sleep(500); |
| } catch (InterruptedException e) {} |
| |
| for (int i=0; i < 200; i++) { |
| try { |
| Thread.sleep(rand.nextInt(50)+1); |
| } catch (InterruptedException e) {} |
| |
| int ds = docsSent.get(); |
| if (ds > 0) { |
| int docToDelete = rand.nextInt(ds) + 1; |
| if (!deletedDocs.contains(docToDelete)) { |
| delI(String.valueOf(docToDelete)); |
| deletedDocs.add(docToDelete); |
| } |
| } |
| } |
| } |
| }; |
| |
| Thread committerThread = new Thread() { |
| public void run() { |
| try { |
| Thread.sleep(rand.nextInt(200)+1); |
| } catch (InterruptedException e) {} |
| |
| for (int i=0; i < 20; i++) { |
| try { |
| cluster.getSolrClient().commit(COLLECTION); |
| } catch (Exception e) {} |
| |
| try { |
| Thread.sleep(rand.nextInt(100)+100); |
| } catch (InterruptedException e) {} |
| } |
| } |
| }; |
| |
| |
| docSenderThread.start(); |
| reloaderThread.start(); |
| committerThread.start(); |
| deleteThread.start(); |
| |
| docSenderThread.join(); |
| reloaderThread.join(); |
| committerThread.join(); |
| deleteThread.join(); |
| |
| cluster.getSolrClient().commit(COLLECTION); |
| |
| if (log.isInfoEnabled()) { |
| log.info("Total of {} docs deleted", deletedDocs.size()); |
| } |
| |
| maxOnLeader = getMaxVersionFromIndex(leader); |
| maxOnReplica = getMaxVersionFromIndex(replica); |
| assertEquals("leader and replica should have same max version before reload", maxOnLeader, maxOnReplica); |
| |
| reloadCollection(leader, COLLECTION); |
| |
| maxOnLeader = getMaxVersionFromIndex(leader); |
| maxOnReplica = getMaxVersionFromIndex(replica); |
| assertEquals("leader and replica should have same max version after reload", maxOnLeader, maxOnReplica); |
| |
| assertDocsExistInAllReplicas(leader, notLeaders, COLLECTION, 1, 1000, deletedDocs); |
| |
| } |
| |
| protected long getMaxVersionFromIndex(Replica replica) throws IOException, SolrServerException { |
| return getVersionFromIndex(replica, null); |
| } |
| |
| protected long getVersionFromIndex(Replica replica, String docId) throws IOException, SolrServerException { |
| Long vers = null; |
| String queryStr = (docId != null) ? "id:" + docId : "_version_:[0 TO *]"; |
| SolrQuery query = new SolrQuery(queryStr); |
| query.setRows(1); |
| query.setFields("id", "_version_"); |
| query.addSort(new SolrQuery.SortClause("_version_", SolrQuery.ORDER.desc)); |
| query.setParam("distrib", false); |
| |
| try (SolrClient client = getHttpSolrClient(replica.getCoreUrl())) { |
| QueryResponse qr = client.query(query); |
| SolrDocumentList hits = qr.getResults(); |
| if (hits.isEmpty()) |
| fail("No results returned from query: "+query); |
| |
| vers = (Long) hits.get(0).getFirstValue("_version_"); |
| } |
| |
| if (vers == null) |
| fail("Failed to get version using query " + query + " from " + replica.getCoreUrl()); |
| |
| return vers.longValue(); |
| } |
| |
| protected void assertDocsExistInAllReplicas(Replica leader, List<Replica> notLeaders, |
| String testCollectionName, |
| int firstDocId, |
| int lastDocId, |
| Set<Integer> deletedDocs) |
| throws Exception { |
| HttpSolrClient leaderSolr = getHttpSolrClient(leader); |
| List<HttpSolrClient> replicas = new ArrayList<HttpSolrClient>(notLeaders.size()); |
| for (Replica r : notLeaders) |
| replicas.add(getHttpSolrClient(r)); |
| |
| try { |
| for (int d = firstDocId; d <= lastDocId; d++) { |
| |
| if (deletedDocs != null && deletedDocs.contains(d)) |
| continue; |
| |
| String docId = String.valueOf(d); |
| Long leaderVers = assertDocExists(leaderSolr, testCollectionName, docId, null); |
| for (HttpSolrClient replicaSolr : replicas) |
| assertDocExists(replicaSolr, testCollectionName, docId, leaderVers); |
| } |
| } finally { |
| if (leaderSolr != null) { |
| leaderSolr.close(); |
| } |
| for (HttpSolrClient replicaSolr : replicas) { |
| replicaSolr.close(); |
| } |
| } |
| } |
| |
| protected HttpSolrClient getHttpSolrClient(Replica replica) throws Exception { |
| return getHttpSolrClient(replica.getCoreUrl()); |
| } |
| |
| protected void sendDoc(int docId) throws Exception { |
| SolrInputDocument doc = new SolrInputDocument(); |
| doc.addField("id", String.valueOf(docId)); |
| doc.addField("a_t", "hello" + docId); |
| AbstractFullDistribZkTestBase.sendDocsWithRetry(cluster.getSolrClient(), COLLECTION, Collections.singletonList(doc), 2, 3, 100); |
| } |
| |
| /** |
| * Query the real-time get handler for a specific doc by ID to verify it |
| * exists in the provided server, using distrib=false so it doesn't route to another replica. |
| */ |
| @SuppressWarnings("rawtypes") |
| protected Long assertDocExists(HttpSolrClient solr, String coll, String docId, Long expVers) throws Exception { |
| QueryRequest qr = new QueryRequest(params("qt", "/get", "id", docId, "distrib", "false", "fl", "id,_version_")); |
| NamedList rsp = solr.request(qr); |
| SolrDocument doc = (SolrDocument)rsp.get("doc"); |
| String match = JSONTestUtil.matchObj("/id", doc, docId); |
| assertTrue("Doc with id=" + docId + " not found in " + solr.getBaseURL() + |
| " due to: " + match + "; rsp=" + rsp, match == null); |
| |
| Long vers = (Long)doc.getFirstValue("_version_"); |
| assertNotNull(vers); |
| if (expVers != null) |
| assertEquals("expected version of doc "+docId+" to be "+expVers, expVers, vers); |
| |
| return vers; |
| } |
| |
| protected boolean reloadCollection(Replica replica, String testCollectionName) throws Exception { |
| ZkCoreNodeProps coreProps = new ZkCoreNodeProps(replica); |
| String coreName = coreProps.getCoreName(); |
| boolean reloadedOk = false; |
| try (HttpSolrClient client = getHttpSolrClient(coreProps.getBaseUrl())) { |
| CoreAdminResponse statusResp = CoreAdminRequest.getStatus(coreName, client); |
| long leaderCoreStartTime = statusResp.getStartTime(coreName).getTime(); |
| |
| Thread.sleep(1000); |
| |
| // send reload command for the collection |
| log.info("Sending RELOAD command for {}", testCollectionName); |
| CollectionAdminRequest.reloadCollection(testCollectionName) |
| .process(client); |
| Thread.sleep(2000); // reload can take a short while |
| |
| // verify reload is done, waiting up to 30 seconds for slow test environments |
| long timeout = System.nanoTime() + TimeUnit.NANOSECONDS.convert(30, TimeUnit.SECONDS); |
| while (System.nanoTime() < timeout) { |
| statusResp = CoreAdminRequest.getStatus(coreName, client); |
| long startTimeAfterReload = statusResp.getStartTime(coreName).getTime(); |
| if (startTimeAfterReload > leaderCoreStartTime) { |
| reloadedOk = true; |
| break; |
| } |
| // else ... still waiting to see the reloaded core report a later start time |
| Thread.sleep(1000); |
| } |
| } |
| return reloadedOk; |
| } |
| } |
