Java > Open Source Codes > org > alfresco > repo > node > index > FullIndexRecoveryComponent


1 /*
2  * Copyright (C) 2006 Alfresco, Inc.
3  *
4  * Licensed under the Mozilla Public License version 1.1
5  * with a permitted attribution clause. You may obtain a
6  * copy of the License at
7  *
8  * http://www.alfresco.org/legal/license.txt
9  *
10  * Unless required by applicable law or agreed to in writing,
11  * software distributed under the License is distributed on an
12  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
13  * either express or implied. See the License for the specific
14  * language governing permissions and limitations under the
15  * License.
16  */
17 package org.alfresco.repo.node.index;
18
19 import java.util.ArrayList ;
20 import java.util.List ;
21
22 import org.alfresco.error.AlfrescoRuntimeException;
23 import org.alfresco.model.ContentModel;
24 import org.alfresco.repo.domain.NodeStatus;
25 import org.alfresco.repo.search.Indexer;
26 import org.alfresco.repo.search.impl.lucene.LuceneIndexerImpl;
27 import org.alfresco.repo.search.impl.lucene.fts.FullTextSearchIndexer;
28 import org.alfresco.repo.transaction.TransactionUtil;
29 import org.alfresco.repo.transaction.TransactionUtil.TransactionWork;
30 import org.alfresco.service.cmr.repository.ChildAssociationRef;
31 import org.alfresco.service.cmr.repository.NodeRef;
32 import org.alfresco.service.cmr.repository.NodeService;
33 import org.alfresco.service.cmr.repository.StoreRef;
34 import org.alfresco.service.cmr.search.ResultSet;
35 import org.alfresco.service.cmr.search.SearchParameters;
36 import org.alfresco.service.cmr.search.SearchService;
37 import org.alfresco.service.transaction.TransactionService;
38 import org.apache.commons.logging.Log;
39 import org.apache.commons.logging.LogFactory;
40 import org.hibernate.CacheMode;
41 import org.hibernate.Query;
42 import org.hibernate.Session;
43 import org.springframework.orm.hibernate3.HibernateCallback;
44 import org.springframework.orm.hibernate3.support.HibernateDaoSupport;
45
46 /**
47  * Ensures that the FTS indexing picks up on any outstanding documents that
48  * require indexing.
49  * <p>
50  * This component must be used as a singleton (one per VM) and may only be
51  * called to reindex once. It will start a thread that processes all available
52  * transactions and keeps checking to ensure that the index is up to date with
53  * the latest database changes.
54  * <p>
55  * <b>The following points are important:</b>
56  * <ul>
57  * <li>
58  * By default, the Hibernate L2 cache is used during processing.
59  * This can be disabled by either disabling the L2 cache globally
60  * for the server (not recommended) or by setting the
61  * {@link #setL2CacheMode(String) l2CacheMode} property. If the
62  * database is static then the L2 cache usage can be set to use
63  * the <code>NORMAL</code> mode. <code>REFRESH</code> should be
64  * used where the server will still be accessed from some clients
65  * despite the database changing.
66  * </li>
67  * <li>
68  * This process should not run continuously on a live
69  * server as it would be performing unecessary work.
70  * If it was left running, however, it would not
71  * lead to data corruption or such-like. Use the
72  * {@link #setRunContinuously(boolean) runContinuously} property
73  * to change this behaviour.
74  * </li>
75  * </ul>
76  *
77  * @author Derek Hulley
78  */
79 public class FullIndexRecoveryComponent extends HibernateDaoSupport implements IndexRecovery
80 {
81     public static final String QUERY_GET_NEXT_CHANGE_TXN_IDS = "node.GetNextChangeTxnIds";
82     public static final String QUERY_GET_CHANGED_NODE_STATUSES = "node.GetChangedNodeStatuses";
83     public static final String QUERY_GET_CHANGED_NODE_STATUSES_COUNT = "node.GetChangedNodeStatusesCount";
84     
85     private static final String START_TXN_ID = "000";
86     
87     private static Log logger = LogFactory.getLog(FullIndexRecoveryComponent.class);
88     
89     /** ensures that this process is kicked off once per VM */
90     private static boolean started = false;
91     /** The current transaction ID being processed */
92     private static String currentTxnId = START_TXN_ID;
93     /** kept to notify the thread that it should quite */
94     private boolean killThread = false;
95     
96     /** provides transactions to atomically index each missed transaction */
97     private TransactionService transactionService;
98     /** the component to index the node hierarchy */
99     private Indexer indexer;
100     /** the FTS indexer that we will prompt to pick up on any un-indexed text */
101     private FullTextSearchIndexer ftsIndexer;
102     /** the component providing searches of the indexed nodes */
103     private SearchService searcher;
104     /** the component giving direct access to <b>node</b> instances */
105     private NodeService nodeService;
106     /** the stores to reindex */
107     private List <StoreRef> storeRefs;
108     /** set this to run the index recovery component */
109     private boolean executeFullRecovery;
110     /** set this on to keep checking for new transactions and never stop */
111     private boolean runContinuously;
112     /** set the time to wait between checking indexes */
113     private long waitTime;
114     /** controls how the L2 cache is used */
115     private CacheMode l2CacheMode;
116     
117     /**
118      * @return Returns the ID of the current (or last) transaction processed
119      */
120     public static String getCurrentTransactionId()
121     {
122         return currentTxnId;
123     }
124
125     public FullIndexRecoveryComponent()
126     {
127         this.storeRefs = new ArrayList <StoreRef>(2);
128         
129         this.killThread = false;
130         this.executeFullRecovery = false;
131         this.runContinuously = false;
132         this.waitTime = 1000L;
133         this.l2CacheMode = CacheMode.REFRESH;
134
135         // ensure that we kill the thread when the VM is shutting down
136 Runnable shutdownRunnable = new Runnable ()
137         {
138             public void run()
139             {
140                 killThread = true;
141             };
142         };
143         Thread shutdownThread = new Thread (shutdownRunnable);
144         Runtime.getRuntime().addShutdownHook(shutdownThread);
145     }
146
147     /**
148      * @return Returns true if the component has already been started
149      */
150     public static boolean isStarted()
151     {
152         return started;
153     }
154
155     /**
156      * @param transactionService provide transactions to index each missed transaction
157      */
158     public void setTransactionService(TransactionService transactionService)
159     {
160         this.transactionService = transactionService;
161     }
162
163     /**
164      * @param indexer the indexer that will be index
165      */
166     public void setIndexer(Indexer indexer)
167     {
168         this.indexer = indexer;
169     }
170     
171     /**
172      * @param ftsIndexer the FTS background indexer
173      */
174     public void setFtsIndexer(FullTextSearchIndexer ftsIndexer)
175     {
176         this.ftsIndexer = ftsIndexer;
177     }
178
179     /**
180      * @param searcher component providing index searches
181      */
182     public void setSearcher(SearchService searcher)
183     {
184         this.searcher = searcher;
185     }
186
187     /**
188      * @param nodeService provides information about nodes for indexing
189      */
190     public void setNodeService(NodeService nodeService)
191     {
192         this.nodeService = nodeService;
193     }
194     
195     /**
196      * Set the stores that need reindexing
197      *
198      * @param storeRefStrings a list of strings representing store references
199      */
200     public void setStores(List <String > storeRefStrings)
201     {
202         storeRefs.clear();
203         for (String storeRefStr : storeRefStrings)
204         {
205             StoreRef storeRef = new StoreRef(storeRefStr);
206             storeRefs.add(storeRef);
207         }
208     }
209
210     /**
211      * Set this to <code>true</code> to initiate the full index recovery.
212      * <p>
213      * This used to default to <code>true</code> but is now false. Set this
214      * if the potentially long-running process of checking and fixing the
215      * indexes must be started.
216      *
217      * @param executeFullRecovery
218      */
219     public void setExecuteFullRecovery(boolean executeFullRecovery)
220     {
221         this.executeFullRecovery = executeFullRecovery;
222     }
223
224     /**
225      * Set this to ensure that the process continuously checks for new transactions.
226      * If not, it will permanently terminate once it catches up with the current
227      * transactions.
228      *
229      * @param runContinuously true to never cease looking for new transactions
230      */
231     public void setRunContinuously(boolean runContinuously)
232     {
233         this.runContinuously = runContinuously;
234     }
235
236     /**
237      * Set the time to wait between checking for new transaction changes in the database.
238      *
239      * @param waitTime the time to wait in milliseconds
240      */
241     public void setWaitTime(long waitTime)
242     {
243         this.waitTime = waitTime;
244     }
245
246     /**
247      * Set the hibernate cache mode by name
248      *
249      * @see org.hibernate.CacheMode
250      */
251     public void setL2CacheMode(String l2CacheModeStr)
252     {
253         if (l2CacheModeStr.equals("GET"))
254         {
255             l2CacheMode = CacheMode.GET;
256         }
257         else if (l2CacheModeStr.equals("IGNORE"))
258         {
259             l2CacheMode = CacheMode.IGNORE;
260         }
261         else if (l2CacheModeStr.equals("NORMAL"))
262         {
263             l2CacheMode = CacheMode.NORMAL;
264         }
265         else if (l2CacheModeStr.equals("PUT"))
266         {
267             l2CacheMode = CacheMode.PUT;
268         }
269         else if (l2CacheModeStr.equals("REFRESH"))
270         {
271             l2CacheMode = CacheMode.REFRESH;
272         }
273         else
274         {
275             throw new IllegalArgumentException ("Unrecognised Hibernate L2 cache mode: " + l2CacheModeStr);
276         }
277     }
278
279     /**
280      * Ensure that the index is up to date with the current state of the persistence layer.
281      * The full list of unique transaction change IDs is retrieved and used to detect
282      * which are not present in the index. All the node changes and deletions for the
283      * remaining transactions are then indexed.
284      */
285     public synchronized void reindex()
286     {
287         if (FullIndexRecoveryComponent.started)
288         {
289             throw new AlfrescoRuntimeException
290                     ("Only one FullIndexRecoveryComponent may be used per VM and it may only be called once");
291         }
292         
293         // ensure that we don't redo this work
294 FullIndexRecoveryComponent.started = true;
295         
296         // work to mark the stores for full text reindexing
297 TransactionWork<Object > ftsReindexWork = new TransactionWork<Object >()
298         {
299             public Object doWork()
300             {
301                 // reindex each store
302 for (StoreRef storeRef : storeRefs)
303                 {
304                     // check if the store exists
305 if (!nodeService.exists(storeRef))
306                     {
307                         // store does not exist
308 if (logger.isDebugEnabled())
309                         {
310                             logger.debug("Skipping reindex of non-existent store: " + storeRef);
311                         }
312                         continue;
313                     }
314                     
315                     // prompt FTS to reindex the store
316 ftsIndexer.requiresIndex(storeRef);
317                 }
318                 // done
319 if (logger.isDebugEnabled())
320                 {
321                     logger.debug("Prompted FTS index on stores: " + storeRefs);
322                 }
323                 return null;
324             }
325         };
326         TransactionUtil.executeInNonPropagatingUserTransaction(transactionService, ftsReindexWork);
327
328         // start full index recovery, if necessary
329 if (!this.executeFullRecovery)
330         {
331             if (logger.isDebugEnabled())
332             {
333                 logger.debug("Full index recovery is off - quitting");
334             }
335         }
336         else
337         {
338             // set the state of the reindex
339 FullIndexRecoveryComponent.currentTxnId = START_TXN_ID;
340             
341             // start a stateful thread that will begin processing the reindexing the transactions
342 Runnable runnable = new ReindexRunner();
343             Thread reindexThread = new Thread (runnable);
344             // make it a daemon thread
345 reindexThread.setDaemon(true);
346             // it should not be a high priority
347 reindexThread.setPriority(Thread.MIN_PRIORITY);
348             // start it
349 reindexThread.start();
350             
351             if (logger.isDebugEnabled())
352             {
353                 logger.debug("Full index recovery thread started: \n" +
354                         " continuous: " + runContinuously + "\n" +
355                         " stores: " + storeRefs);
356             }
357         }
358     }
359     
360     /**
361      * Stateful thread runnable that executes reindex calls.
362      *
363      * @see FullIndexRecoveryComponent#reindexNodes()
364      *
365      * @author Derek Hulley
366      */
367     private class ReindexRunner implements Runnable
368     {
369         public void run()
370         {
371             // keep this thread going permanently
372 while (!killThread)
373             {
374                 try
375                 {
376                     // reindex nodes
377 List <String > txnsIndexed = FullIndexRecoveryComponent.this.reindexNodes();
378                     // reindex missing content
379 int missingContentCount = FullIndexRecoveryComponent.this.reindexMissingContent();
380                     // check if the process should terminate
381 if (txnsIndexed.size() == 0 && !runContinuously)
382                     {
383                         // the thread has caught up with all the available work and should not
384 // run continuously
385 if (logger.isDebugEnabled())
386                         {
387                             logger.debug("Thread quitting - no more available indexing to do: \n" +
388                                     " last txn: " + FullIndexRecoveryComponent.getCurrentTransactionId());
389                         }
390                         break;
391                     }
392                     // brief pause
393 synchronized(FullIndexRecoveryComponent.this)
394                     {
395                         FullIndexRecoveryComponent.this.wait(waitTime);
396                     }
397                 }
398                 catch (InterruptedException e)
399                 {
400                     // ignore
401 }
402                 catch (Throwable e)
403                 {
404                     if (killThread)
405                     {
406                         // the shutdown may have caused the exception - ignore it
407 }
408                     else
409                     {
410                         // we are still a go; report it
411 logger.error("Reindex failure", e);
412                     }
413                 }
414             }
415         }
416     }
417
418     /**
419      * @return Returns the number of documents reindexed
420      */
421     private int reindexMissingContent()
422     {
423         int count = 0;
424         for (StoreRef storeRef : storeRefs)
425         {
426             count += reindexMissingContent(storeRef);
427         }
428         return count;
429     }
430     
431     /**
432      * @param storeRef the store to check for missing content
433      * @return Returns the number of documents reindexed
434      */
435     private int reindexMissingContent(StoreRef storeRef)
436     {
437         SearchParameters sp = new SearchParameters();
438         sp.addStore(storeRef);
439
440         // search for it in the index
441 String query = "TEXT:" + LuceneIndexerImpl.NOT_INDEXED_CONTENT_MISSING;
442         sp.setLanguage(SearchService.LANGUAGE_LUCENE);
443         sp.setQuery(query);
444         ResultSet results = null;
445         try
446         {
447             results = searcher.query(sp);
448             
449             int count = 0;
450             // loop over the results and get the details of the nodes that have missing content
451 List <ChildAssociationRef> assocRefs = results.getChildAssocRefs();
452             for (ChildAssociationRef assocRef : assocRefs)
453             {
454                 final NodeRef childNodeRef = assocRef.getChildRef();
455                 // prompt for a reindex - it might fail again, but we just keep plugging away
456 TransactionWork<Object > reindexWork = new TransactionWork<Object >()
457                 {
458                     public Object doWork()
459                     {
460                         indexer.updateNode(childNodeRef);
461                         return null;
462                     }
463                 };
464                 TransactionUtil.executeInNonPropagatingUserTransaction(transactionService, reindexWork);
465                 count++;
466             }
467             // done
468 if (logger.isDebugEnabled())
469             {
470                 logger.debug("Reindexed missing content: \n" +
471                         " store: " + storeRef + "\n" +
472                         " node count: " + count);
473             }
474             return count;
475         }
476         finally
477         {
478             if (results != null)
479             {
480                 results.close();
481             }
482         }
483     }
484     
485     /**
486      * @return Returns the transaction ID just reindexed, i.e. where some work was performed
487      */
488     private List <String > reindexNodes()
489     {
490         // get a list of all transactions still requiring a check
491 List <String > txnsToCheck = getNextChangeTxnIds(FullIndexRecoveryComponent.currentTxnId);
492         
493         // loop over each transaction
494 for (String changeTxnId : txnsToCheck)
495         {
496             reindexNodes(changeTxnId);
497         }
498         
499         // done
500 return txnsToCheck;
501     }
502     
503     /**
504      * Reindexes changes specific to the change transaction ID.
505      * <p>
506      * <b>All exceptions are absorbed.</b>
507      */
508     private void reindexNodes(final String changeTxnId)
509     {
510         /*
511          * This must execute each within its own transaction.
512          * The cache size is therefore not an issue.
513          */
514         TransactionWork<Object > reindexWork = new TransactionWork<Object >()
515         {
516             public Object doWork() throws Exception
517             {
518                 // perform the work in a Hibernate callback
519 HibernateCallback callback = new ReindexCallback(changeTxnId);
520                 getHibernateTemplate().execute(callback);
521                 // done
522 return null;
523             }
524         };
525         try
526         {
527             TransactionUtil.executeInNonPropagatingUserTransaction(transactionService, reindexWork);
528         }
529         catch (Throwable e)
530         {
531             logger.error("Transaction reindex failed: \n" +
532                     " txn: " + changeTxnId,
533                     e);
534         }
535         finally
536         {
537             // Up the current transaction now, in case the process fails at this point.
538 // This will prevent the transaction from being processed again.
539 // This applies to failures as well, which should be dealt with externally
540 // and having the entire process start again, e.g. such as a system reboot
541 currentTxnId = changeTxnId;
542         }
543     }
544     
545     /**
546      * Stateful inner class that implements a single reindex call for a given store
547      * and transaction.
548      * <p>
549      * It must be called within its own transaction.
550      *
551      * @author Derek Hulley
552      */
553     private class ReindexCallback implements HibernateCallback
554     {
555         private final String changeTxnId;
556         
557         public ReindexCallback(String changeTxnId)
558         {
559             this.changeTxnId = changeTxnId;
560         }
561         
562         /**
563          * Changes the L2 cache usage before reindexing for each store
564          *
565          * @see #reindexNodes(StoreRef, String)
566          */
567         public Object doInHibernate(Session session)
568         {
569             // set the way the L2 cache is used
570 getSession().setCacheMode(l2CacheMode);
571             
572             // reindex each store
573 for (StoreRef storeRef : storeRefs)
574             {
575                 if (!nodeService.exists(storeRef))
576                 {
577                     // the store is not present
578 continue;
579                 }
580                 // reindex for store
581 reindexNodes(storeRef, changeTxnId);
582             }
583             // done
584 return null;
585         }
586         
587         private void reindexNodes(StoreRef storeRef, String changeTxnId)
588         {
589             // check if we need to perform this operation
590 SearchParameters sp = new SearchParameters();
591             sp.addStore(storeRef);
592
593             // search for it in the index
594 String query = "TX:\"" + changeTxnId + "\"";
595             sp.setLanguage(SearchService.LANGUAGE_LUCENE);
596             sp.setQuery(query);
597             ResultSet results = null;
598             try
599             {
600                 results = searcher.query(sp);
601                 // did the index have any of these changes?
602 if (results.length() > 0)
603                 {
604                     // the transaction has an entry in the index - assume that it was
605 // atomically correct
606 if (logger.isDebugEnabled())
607                     {
608                         logger.debug("Transaction present in index - no indexing required: \n" +
609                                 " store: " + storeRef + "\n" +
610                                 " txn: " + changeTxnId);
611                     }
612                     return;
613                 }
614             }
615             finally
616             {
617                 if (results != null)
618                 {
619                     results.close();
620                 }
621             }
622             // the index has no record of this
623 // were there any changes, or is it all just deletions?
624 int changedCount = getChangedNodeStatusesCount(storeRef, changeTxnId);
625             if (changedCount == 0)
626             {
627                 // no nodes were changed in the transaction, i.e. they are only deletions
628 // the index is quite right not to have any entries for the transaction
629 if (logger.isDebugEnabled())
630                 {
631                     logger.debug("Transaction only has deletions - no indexing required: \n" +
632                             " store: " + storeRef + "\n" +
633                             " txn: " + changeTxnId);
634                 }
635                 return;
636             }
637
638             // process the deletions relevant to the txn and the store
639 List <NodeStatus> deletedNodeStatuses = getDeletedNodeStatuses(storeRef, changeTxnId);
640             for (NodeStatus status : deletedNodeStatuses)
641             {
642                 NodeRef nodeRef = new NodeRef(storeRef, status.getKey().getGuid());
643                 // only the child node ref is relevant
644 ChildAssociationRef assocRef = new ChildAssociationRef(
645                         ContentModel.ASSOC_CHILDREN,
646                         null,
647                         null,
648                         nodeRef);
649                 indexer.deleteNode(assocRef);
650             }
651             
652             // process additions
653 List <NodeStatus> changedNodeStatuses = getChangedNodeStatuses(storeRef, changeTxnId);
654             for (NodeStatus status : changedNodeStatuses)
655             {
656                 NodeRef nodeRef = new NodeRef(storeRef, status.getKey().getGuid());
657                 // get the primary assoc for the node
658 ChildAssociationRef primaryAssocRef = nodeService.getPrimaryParent(nodeRef);
659                 // reindex
660 indexer.createNode(primaryAssocRef);
661             }
662             
663             // done
664 if (logger.isDebugEnabled())
665             {
666                 logger.debug("Transaction reindexed: \n" +
667                         " store: " + storeRef + "\n" +
668                         " txn: " + changeTxnId + "\n" +
669                         " deletions: " + deletedNodeStatuses.size() + "\n" +
670                         " modifications: " + changedNodeStatuses.size());
671             }
672         }
673     };
674
675     /**
676      * Retrieve all transaction IDs that are greater than the given transaction ID.
677      *
678      * @param currentTxnId the transaction ID that must be less than all returned results
679      * @return Returns an ordered list of transaction IDs
680      */
681     @SuppressWarnings ("unchecked")
682     public List <String > getNextChangeTxnIds(final String currentTxnId)
683     {
684         HibernateCallback callback = new HibernateCallback()
685         {
686             public Object doInHibernate(Session session)
687             {
688                 Query query = session.getNamedQuery(QUERY_GET_NEXT_CHANGE_TXN_IDS);
689                 query.setString("currentTxnId", currentTxnId)
690                      .setReadOnly(true);
691                 return query.list();
692             }
693         };
694         List <String > queryResults = (List <String >) getHibernateTemplate().execute(callback);
695         // done
696 return queryResults;
697     }
698
699     @SuppressWarnings ("unchecked")
700     public int getChangedNodeStatusesCount(final StoreRef storeRef, final String changeTxnId)
701     {
702         HibernateCallback callback = new HibernateCallback()
703         {
704             public Object doInHibernate(Session session)
705             {
706                 Query query = session.getNamedQuery(QUERY_GET_CHANGED_NODE_STATUSES_COUNT);
707                 query.setBoolean("deleted", false)
708                      .setString("storeProtocol", storeRef.getProtocol())
709                      .setString("storeIdentifier", storeRef.getIdentifier())
710                      .setString("changeTxnId", changeTxnId)
711                      .setReadOnly(true);
712                 return query.uniqueResult();
713             }
714         };
715         Integer changeCount = (Integer ) getHibernateTemplate().execute(callback);
716         // done
717 return changeCount.intValue();
718     }
719
720     @SuppressWarnings ("unchecked")
721     public List <NodeStatus> getChangedNodeStatuses(final StoreRef storeRef, final String changeTxnId)
722     {
723         HibernateCallback callback = new HibernateCallback()
724         {
725             public Object doInHibernate(Session session)
726             {
727                 Query query = session.getNamedQuery(QUERY_GET_CHANGED_NODE_STATUSES);
728                 query.setBoolean("deleted", false)
729                      .setString("storeProtocol", storeRef.getProtocol())
730                      .setString("storeIdentifier", storeRef.getIdentifier())
731                      .setString("changeTxnId", changeTxnId)
732                      .setReadOnly(true);
733                 return query.list();
734             }
735         };
736         List <NodeStatus> queryResults = (List ) getHibernateTemplate().execute(callback);
737         // done
738 return queryResults;
739     }
740
741     @SuppressWarnings ("unchecked")
742     public List <NodeStatus> getDeletedNodeStatuses(final StoreRef storeRef, final String changeTxnId)
743     {
744         HibernateCallback callback = new HibernateCallback()
745         {
746             public Object doInHibernate(Session session)
747             {
748                 Query query = session.getNamedQuery(QUERY_GET_CHANGED_NODE_STATUSES);
749                 query.setBoolean("deleted", true)
750                      .setString("storeProtocol", storeRef.getProtocol())
751                      .setString("storeIdentifier", storeRef.getIdentifier())
752                      .setString("changeTxnId", changeTxnId)
753                      .setReadOnly(true);
754                 return query.list();
755             }
756         };
757         List <NodeStatus> queryResults = (List ) getHibernateTemplate().execute(callback);
758         // done
759 return queryResults;
760     }
761 }

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags