SEARCH-506: Adding an explicit shard router

2026-09-16 18:12:56 +00:00 · 2017-08-08 13:36:34 +02:00
parent 67cc26c635
commit 1bbcd35484
3 changed files with 178 additions and 0 deletions
@@ -19,6 +19,8 @@
 package org.alfresco.solr.tracker;

 import org.alfresco.repo.index.shard.ShardMethodEnum;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;

 import java.util.Properties;
 import java.util.Set;
@@ -29,26 +31,49 @@ import java.util.Set;

 public class DocRouterFactory
 {
+    protected final static Logger log = LoggerFactory.getLogger(DocRouterFactory.class);
+
    public static DocRouter getRouter(Properties properties, ShardMethodEnum method) {

+        String shardDotId = properties.getProperty("shard.id");
+        if (shardDotId != null  && !shardDotId.isEmpty())
+        {
+            try
+            {
+                int shardid = Integer.parseInt(shardDotId);
+                log.info("Sharding via an ExplicitRouter for shard "+shardid);
+                return new ExplicitRouter(shardid);
+            } catch (NumberFormatException e)
+            {
+                log.error("Failed to parse a shard.id of "+shardDotId);
+            }
+        }
+
        switch(method) {
            case DB_ID:
+                log.info("Sharding via DB_ID");
                return new DBIDRouter();
            case DB_ID_RANGE:
                String range = properties.getProperty("shard.range");
                String[] rangeParts = range.split("-");
                long startRange = Long.parseLong(rangeParts[0].trim());
                long endRange = Long.parseLong(rangeParts[1].trim());
+                log.info("Sharding via DB_ID_RANGE");
                return new DBIDRangeRouter(startRange, endRange);
            case ACL_ID:
+                log.info("Sharding via ACL_ID");
                return new ACLIDMurmurRouter();
            case MOD_ACL_ID:
+                log.info("Sharding via MOD_ACL_ID");
                return new ACLIDModRouter();
            case DATE:
+                log.info("Sharding via DATE");
                return new DateMonthRouter(properties.getProperty("shard.date.grouping", "1"));
            case PROPERTY:
+                log.info("Sharding via PROPERTY");
                return new PropertyRouter(properties.getProperty("shard.regex", ""));
            default:
+                log.info("Sharding via DB_ID (default)");
                return new DBIDRouter();
        }
    }
@@ -0,0 +1,26 @@
+package org.alfresco.solr.tracker;
+
+import org.alfresco.solr.client.Acl;
+import org.alfresco.solr.client.Node;
+
+/**
+ * Routes a document only if the shardInstance matches the provided shardId
+ */
+public class ExplicitRouter implements DocRouter {
+
+    private final int shardId;
+
+    public ExplicitRouter(int shardId) {
+        this.shardId = shardId;
+    }
+
+    @Override
+    public boolean routeAcl(int shardCount, int shardInstance, Acl acl) {
+        return shardId == shardInstance;
+    }
+
+    @Override
+    public boolean routeNode(int shardCount, int shardInstance, Node node) {
+        return shardId == shardInstance;
+    }
+}
@@ -0,0 +1,127 @@
+/*
+ * Copyright (C) 2005-2016 Alfresco Software Limited.
+ *
+ * This file is part of Alfresco
+ *
+ * Alfresco is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Alfresco is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with Alfresco. If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.alfresco.solr.tracker;
+
+import com.carrotsearch.randomizedtesting.RandomizedContext;
+import org.alfresco.model.ContentModel;
+import org.alfresco.repo.index.shard.ShardMethodEnum;
+import org.alfresco.solr.AbstractAlfrescoDistributedTest;
+import org.alfresco.solr.SolrInformationServer;
+import org.alfresco.solr.client.*;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.TermQuery;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.core.SolrCore;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.*;
+
+import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_DOC_TYPE;
+import static org.alfresco.solr.AlfrescoSolrUtils.*;
+
+/**
+ * Test Routes based on an explicit shard
+ *
+ * @author Gethin James
+ */
+@SolrTestCaseJ4.SuppressSSL
+@SolrTestCaseJ4.SuppressObjectReleaseTracker (bugUrl = "RAMDirectory")
+@LuceneTestCase.SuppressCodecs({"Appending","Lucene3x","Lucene40","Lucene41","Lucene42","Lucene43", "Lucene44", "Lucene45","Lucene46","Lucene47","Lucene48","Lucene49"})
+public class DistributedExplicitShardRoutingTrackerTest extends AbstractAlfrescoDistributedTest
+{
+    @Rule
+    public JettyServerRule jetty = new JettyServerRule(this.getClass().getSimpleName(), 3, getProperties(), new String[]{DEFAULT_TEST_CORENAME});
+
+    @Test
+    public void testShardId() throws Exception
+    {
+        putHandleDefaults();
+
+        int numAcls = 25;
+        AclChangeSet bulkAclChangeSet = getAclChangeSet(numAcls);
+
+        List<Acl> bulkAcls = new ArrayList();
+        List<AclReaders> bulkAclReaders = new ArrayList();
+
+
+        for (int i = 0; i < numAcls; i++) {
+            Acl bulkAcl = getAcl(bulkAclChangeSet);
+            bulkAcls.add(bulkAcl);
+            bulkAclReaders.add(getAclReaders(bulkAclChangeSet,
+                    bulkAcl,
+                    list("king" + bulkAcl.getId()),
+                    list("king" + bulkAcl.getId()),
+                    null));
+        }
+
+        indexAclChangeSet(bulkAclChangeSet,
+                bulkAcls,
+                bulkAclReaders);
+
+        int numNodes = 1000;
+        List<Node> nodes = new ArrayList();
+        List<NodeMetaData> nodeMetaDatas = new ArrayList();
+
+        Transaction bigTxn = getTransaction(0, numNodes);
+        RandomizedContext context = RandomizedContext.current();
+        Random ints = context.getRandom();
+
+        for (int i = 0; i < numNodes; i++) {
+            int aclIndex = i % numAcls;
+            Node node = getNode(bigTxn, bulkAcls.get(aclIndex), Node.SolrApiNodeStatus.UPDATED);
+            nodes.add(node);
+            NodeMetaData nodeMetaData = getNodeMetaData(node, bigTxn, bulkAcls.get(aclIndex), "king", null, false);
+            nodeMetaDatas.add(nodeMetaData);
+        }
+
+        indexTransaction(bigTxn, nodes, nodeMetaDatas);
+
+        Query contentQuery = new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world"));
+        Query aclQuery = new TermQuery(new Term(FIELD_DOC_TYPE, SolrInformationServer.DOC_TYPE_ACL));
+        List<SolrCore> shards = getJettyCores(jettyShards);
+        long begin = System.currentTimeMillis();
+
+        for (SolrCore core : shards)
+        {
+            if ("shard1".endsWith(core.getName()))
+            {
+                waitForDocCountCore(core, contentQuery, numNodes, 100000, begin);
+                waitForDocCountCore(core, aclQuery, numAcls, 100000, begin);
+            }
+        }
+
+        //Now we know that shard1 has indexed nodes, lets make sure the other nodes don't have any.
+        assertShardCount(0, contentQuery, 0);
+        assertShardCount(2, contentQuery, 0);
+
+        assertShardCount(0, aclQuery, 0);
+        assertShardCount(2, aclQuery, 0);
+    }
+
+    protected Properties getProperties()
+    {
+        Properties prop = new Properties();
+        prop.put("shard.id", "1");
+        return prop;
+    }
+}