inteligr8/SearchServices
1
0
Fork 0
mirror of https://github.com/Alfresco/SearchServices.git synced 2026-09-16 18:12:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

[MNT-23072]

Browse Source 
fix index last transaction error in dbid_range sharding
added integration test
This commit is contained in:
Elia
2022-08-23 17:58:05 +02:00
parent a29aafbafb
commit d111a50252
3 changed files with 82 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java
+4 -8
View File
@@ -94,10 +94,6 @@ public class MetadataTracker extends ActivatableTracker
private final ConcurrentLinkedQueue<Long> nodesToIndex = new ConcurrentLinkedQueue<>();
private final ConcurrentLinkedQueue<Long> nodesToPurge = new ConcurrentLinkedQueue<>();
private final ConcurrentLinkedQueue<String> queriesToReindex = new ConcurrentLinkedQueue<>();
private final boolean isRunningInProduction =
!Boolean.parseBoolean(System.getProperty("alfresco.test", "false"));
private ForkJoinPool forkJoinPool;
// Share run and write locks across all MetadataTracker threads
@@ -193,7 +189,7 @@ public class MetadataTracker extends ActivatableTracker
// In order to apply performance optimizations, checking the availability of Repo Web Scripts is required.
// As these services are available from ACS 6.2
if (checkRepoServicesAvailability && isRunningInProduction)
if (checkRepoServicesAvailability)
{
// Try invoking getNextTxCommitTime service
try
@@ -870,7 +866,7 @@ public class MetadataTracker extends ActivatableTracker
latestTransaction.setCommitTimeMs(transactions.getMaxTxnCommitTime());
latestTransaction.setId(transactions.getMaxTxnId());
if (!isTransactionIndexed(latestTransaction))
if (isTransactionToBeIndexed(latestTransaction))
{
transactions = new Transactions(Collections.singletonList(latestTransaction), transactions.getMaxTxnCommitTime(),
transactions.getMaxTxnId());
@@ -887,7 +883,7 @@ public class MetadataTracker extends ActivatableTracker
return transactions;
}
private boolean isTransactionIndexed(Transaction transaction)
private boolean isTransactionToBeIndexed(Transaction transaction)
{
try
{
@@ -997,7 +993,7 @@ public class MetadataTracker extends ActivatableTracker
final AtomicInteger counterTransaction = new AtomicInteger();
Collection<List<Transaction>> txBatches = transactions.getTransactions().stream()
.peek(txnsFound::add)
.filter(this::isTransactionIndexed)
.filter(this::isTransactionToBeIndexed)
.collect(Collectors.groupingBy(transaction -> counterTransaction.getAndAdd(
(int) (transaction.getDeletes() + transaction.getUpdates())) / transactionDocsBatchSize))
.values();
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedDbidRangeAlfrescoSolrTrackerIT.java
+58 -9
View File
@@ -37,6 +37,7 @@ import org.alfresco.solr.client.Transaction;
import org.apache.lucene.index.Term;
import org.apache.lucene.search.TermQuery;
import org.apache.solr.SolrTestCaseJ4;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -47,6 +48,7 @@ import java.util.List;
import java.util.Properties;
import static org.alfresco.repo.search.adaptor.QueryConstants.FIELD_DOC_TYPE;
import static org.alfresco.solr.AlfrescoSolrUtils.MAX_WAIT_TIME;
import static org.alfresco.solr.AlfrescoSolrUtils.getAcl;
import static org.alfresco.solr.AlfrescoSolrUtils.getAclChangeSet;
import static org.alfresco.solr.AlfrescoSolrUtils.getAclReaders;
@@ -65,7 +67,7 @@ public class DistributedDbidRangeAlfrescoSolrTrackerIT extends AbstractAlfrescoD
@BeforeClass
public static void initData() throws Throwable
{
initSolrServers(2, getSimpleClassName(), getShardMethod());
initSolrServers(3, getSimpleClassName(), getShardMethod());
}
@AfterClass
@@ -73,13 +75,14 @@ public class DistributedDbidRangeAlfrescoSolrTrackerIT extends AbstractAlfrescoD
{
dismissSolrServers();
}
@Test
public void testDbIdRange() throws Exception
{
putHandleDefaults();
int numAcls = 250;
@After
public void deleteDataFromIndex() throws Exception {
deleteByQueryAllClients("*:*");
waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 0, MAX_WAIT_TIME);
}
private List<Acl> createAcls(int numAcls){
AclChangeSet bulkAclChangeSet = getAclChangeSet(numAcls);
List<Acl> bulkAcls = new ArrayList<>();
@@ -99,6 +102,17 @@ public class DistributedDbidRangeAlfrescoSolrTrackerIT extends AbstractAlfrescoD
bulkAcls,
bulkAclReaders);
return bulkAcls;
}
@Test
public void testDbIdRange() throws Exception
{
putHandleDefaults();
int numAcls = 250;
var bulkAcls = createAcls(numAcls);
int numNodes = 150;
List<Node> nodes = new ArrayList<>();
List<NodeMetaData> nodeMetaDatas = new ArrayList<>();
@@ -108,14 +122,14 @@ public class DistributedDbidRangeAlfrescoSolrTrackerIT extends AbstractAlfrescoD
for(int i=0; i<numNodes; i++)
{
int aclIndex = i % numAcls;
Node node = getNode((long)i, bigTxn, bulkAcls.get(aclIndex), Node.SolrApiNodeStatus.UPDATED);
Node node = getNode(i, bigTxn, bulkAcls.get(aclIndex), Node.SolrApiNodeStatus.UPDATED);
nodes.add(node);
NodeMetaData nodeMetaData = getNodeMetaData(node, bigTxn, bulkAcls.get(aclIndex), "mike", null, false);
nodeMetaDatas.add(nodeMetaData);
}
indexTransaction(bigTxn, nodes, nodeMetaDatas);
waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), numNodes, 100000);
waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), numNodes, MAX_WAIT_TIME);
waitForDocCountAllCores(new TermQuery(new Term(FIELD_DOC_TYPE, SolrInformationServer.DOC_TYPE_ACL)), numAcls, 80000);
//The test framework has ranges 0-100, 100-200, ...
@@ -123,6 +137,41 @@ public class DistributedDbidRangeAlfrescoSolrTrackerIT extends AbstractAlfrescoD
assertShardCount(1, new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 50);
}
@Test
public void testIndexLastTransaction() throws Exception {
int numAcls = 1;
var acls = createAcls(numAcls);
// first range
Transaction firstRange = getTransaction(0, 1);
firstRange.setId(0);
var node = getNode( 5, firstRange, acls.get(0), Node.SolrApiNodeStatus.UPDATED );
indexTransaction(firstRange, List.of(node), List.of(getNodeMetaData(node, firstRange, acls.get(0), "mike", null, false)));
waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 1, MAX_WAIT_TIME);
// second range
Transaction secondRange = getTransaction(0, 1);
secondRange.setId(1);
node = getNode( 105, firstRange, acls.get(0), Node.SolrApiNodeStatus.UPDATED );
indexTransaction(secondRange, List.of(node), List.of(getNodeMetaData(node, secondRange, acls.get(0), "mike", null, false)));
waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 2, MAX_WAIT_TIME);
for (int i = 0; i< 50; i++){
Transaction trxThirdRange = getTransaction(0, 1);
trxThirdRange.setId(2 + i);
node = getNode( 230 + i, trxThirdRange, acls.get(0), Node.SolrApiNodeStatus.UPDATED );
indexTransaction(trxThirdRange, List.of(node), List.of(getNodeMetaData(node, trxThirdRange, acls.get(0), "mike", null, false)));
}
waitForDocCount(new TermQuery(new Term("content@s___t@{http://www.alfresco.org/model/content/1.0}content", "world")), 52, MAX_WAIT_TIME);
// Check all the shards have indexed the last transaction
assertShardCount(0, new TermQuery(new Term("S_TXID", "51")), 1);
assertShardCount(1, new TermQuery(new Term("S_TXID", "51")), 1);
assertShardCount(2, new TermQuery(new Term("S_TXID", "51")), 1);
}
protected static Properties getShardMethod()
{
Properties prop = new Properties();
search-services/alfresco-solrclient-lib/src/main/java/org/alfresco/solr/client/SOLRAPIQueueClient.java
+20
View File
@@ -40,11 +40,13 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.alfresco.httpclient.Response;
import org.alfresco.repo.dictionary.NamespaceDAO;
import org.alfresco.repo.index.shard.ShardState;
import org.alfresco.service.namespace.QName;
import org.alfresco.util.Pair;
import org.apache.http.HttpStatus;
import org.json.JSONException;
@@ -161,6 +163,24 @@ public class SOLRAPIQueueClient extends SOLRAPIClient
return Collections.emptyList();
}
public Long getNextTxCommitTime(String coreName, Long fromCommitTime) throws NoSuchMethodException {
throw new NoSuchMethodException();
}
public Pair<Long, Long> getTxIntervalCommitTime(String coreName, Long fromNodeId, Long toNodeId)
{
List<Transaction> transactions = TRANSACTION_QUEUE.stream()
.filter(txn -> NODE_MAP.get(txn.getId())
.stream()
.anyMatch(node -> node.getId() >= fromNodeId && node.getId() <= toNodeId))
.collect(Collectors.toList());
if (transactions.size() > 0){
return new Pair<>(transactions.get(0).getCommitTimeMs(), transactions.get(transactions.size() - 1).getCommitTimeMs());
} else {
return new Pair<>(-1l, -1l);
}
}
public Transactions getTransactions(Long fromCommitTime, Long minTxnId, Long toCommitTime, Long maxTxnId, int maxResults) throws IOException, JSONException
{