HSEARCH-323 Fault tolerant FSSlaveDirectoryProvider (Oliver Siegmar)

hibernate-search/src/main/docbook/en-US/modules/configuration.xml

@@ -286,9 +286,20 @@ hibernate.search.Actions.directory_provider com.acme.hibernate.provider.CustomDi
             see <xref
             linkend="search-configuration-directory-lockfactories" />
             </para><para><literal>retry_marker_lookup</literal> : optional,
-            default to 0. Defines how many times, we look for the marker files
+            default to 0. Defines how many times we look for the marker files
             in the source directory before failing. Waiting 5 seconds between
-            each try. </para><para><literal>filesystem_access_type</literal>:
+            each try. </para>
+            <para><literal>retry_initialize_period</literal> : optional,
+            set an integer value in seconds to enable the retry initialize feature:
+            if the slave can't find the master
+            index it will try again until it's found in background, without preventing
+            the application to start: fullText queries performed
+            before the index is initialized are not blocked but will return empty results.
+            When not enabling the option or explicitly setting it to zero it will fail
+            with an exception instead of scheduling a retry timer. To prevent the application from
+            starting without an invalid index but still control an initialization
+            timeout, see <literal>retry_marker_lookup</literal> instead.</para>
+            <para><literal>filesystem_access_type</literal>:
             allows to determine the exact type of
             <classname>FSDirectory</classname> implementation used by this
             <classname>DirectoryProvider</classname>. Allowed values are

hibernate-search/src/main/java/org/hibernate/search/store/DirectoryProviderHelper.java

@@ -30,13 +30,11 @@
 import java.util.Properties;
 
 import org.apache.lucene.analysis.SimpleAnalyzer;
-import org.apache.lucene.index.CorruptIndexException;
 import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.IndexWriter;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
 import org.apache.lucene.store.LockFactory;
-import org.apache.lucene.store.LockObtainFailedException;
 import org.apache.lucene.store.MMapDirectory;
 import org.apache.lucene.store.NIOFSDirectory;
 import org.apache.lucene.store.NativeFSLockFactory;
@@ -48,6 +46,7 @@
 
 import org.hibernate.annotations.common.util.StringHelper;
 import org.hibernate.search.SearchException;
+import org.hibernate.search.backend.configuration.ConfigurationParseHelper;
 import org.hibernate.search.util.ClassLoaderHelper;
 import org.hibernate.search.util.FileHelper;
 import org.hibernate.search.util.LoggerFactory;
@@ -68,6 +67,7 @@ public final class DirectoryProviderHelper {
 	private static final String INDEX_BASE_PROP_NAME = "indexBase";
 	private static final String INDEX_NAME_PROP_NAME = "indexName";
 	private static final String REFRESH_PROP_NAME = "refresh";
+	private static final String RETRY_INITIALIZE_PROP_NAME = "retry_initialize_period";
 
 	private DirectoryProviderHelper() {
 	}
@@ -275,6 +275,20 @@ private static void makeSanityCheckedDirectory(File directory, String indexName,
 			);
 		}
 	}
+
+	/**
+	 * @param properties the configuration of the DirectoryProvider
+	 * @param directoryProviderName the name of the DirectoryProvider, used for error reporting
+	 * @return The period in milliseconds to keep retrying initialization of a DirectoryProvider 
+	 */
+	static long getRetryInitializePeriod(Properties properties, String directoryProviderName) {
+		int retry_period_seconds = ConfigurationParseHelper.getIntValue( properties, RETRY_INITIALIZE_PROP_NAME, 0 );
+		log.debug( "Retry initialize period for Directory {}: {} seconds", directoryProviderName, retry_period_seconds );
+		if ( retry_period_seconds < 0 ) {
+			throw new SearchException( RETRY_INITIALIZE_PROP_NAME + " for Directory " + directoryProviderName + " must be a positive integer" );
+		}
+		return retry_period_seconds * 1000; //convert into milliseconds
+	}
 
 	static long getRefreshPeriod(Properties properties, String directoryProviderName) {
 		String refreshPeriod = properties.getProperty( REFRESH_PROP_NAME, "3600" );

hibernate-search/src/main/java/org/hibernate/search/store/FSSlaveDirectoryProvider.java

@@ -33,11 +33,14 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
+import org.apache.lucene.store.RAMDirectory;
 import org.slf4j.Logger;
 
 import org.hibernate.annotations.common.AssertionFailure;
 import org.hibernate.search.Environment;
+import org.hibernate.search.backend.configuration.ConfigurationParseHelper;
 import org.hibernate.search.spi.BuildContext;
 import org.hibernate.search.SearchException;
 import org.hibernate.search.util.FileHelper;
@@ -54,15 +57,20 @@
  *
  * @author Emmanuel Bernard
  * @author Sanne Grinovero
+ * @author Oliver Siegmar
  */
-public class FSSlaveDirectoryProvider implements DirectoryProvider<FSDirectory> {
+public class FSSlaveDirectoryProvider implements DirectoryProvider<Directory> {
 
 	private static final Logger log = LoggerFactory.make();
 	private final Timer timer = new Timer( true ); //daemon thread, the copy algorithm is robust
 
+	private volatile boolean initialized = false;
+	private volatile boolean started = false;
+
 	private volatile int current; //used also as memory barrier of all other values, which are set once.
 
 	//variables having visibility granted by a read of "current"
+	private volatile Directory dummyDirectory;
 	private FSDirectory directory1;
 	private FSDirectory directory2;
 	private String indexName;
@@ -73,14 +81,13 @@ public class FSSlaveDirectoryProvider implements DirectoryProvider<FSDirectory>
 	private File indexDir;
 	private String directoryProviderName;
 	private Properties properties;
-	private TriggerTask task;
+	private UpdateTask updateTask;
 
 	public void initialize(String directoryProviderName, Properties properties, BuildContext context) {
 		this.properties = properties;
 		this.directoryProviderName = directoryProviderName;
 		//source guessing
 		sourceIndexDir = DirectoryProviderHelper.getSourceDirectory( directoryProviderName, properties, false );
-		checkCurrentMarkerInSource();
 		log.debug( "Source directory: {}", sourceIndexDir.getPath() );
 		indexDir = DirectoryProviderHelper.getVerifiedIndexDir( directoryProviderName, properties, true );
 		log.debug( "Index directory: {}", indexDir.getPath() );
@@ -94,17 +101,11 @@ public void initialize(String directoryProviderName, Properties properties, Buil
 		current = 0; //publish all state to other threads
 	}
 
-	private void checkCurrentMarkerInSource() {
-		int retry;
-		try {
-			retry = Integer.parseInt( properties.getProperty( Environment.RETRY_MARKER_LOOKUP, "0" ) );
-			if (retry < 0) {
-				throw new NumberFormatException( );
-			}
-		}
-		catch ( NumberFormatException e ) {
-			throw new SearchException("retry_marker_lookup options must be a positive number: "
-					+ properties.getProperty( Environment.RETRY_MARKER_LOOKUP ) );
+	private boolean currentMarkerIsInSource() {
+		int retry = ConfigurationParseHelper.getIntValue( properties, Environment.RETRY_MARKER_LOOKUP, 0 );
+		if ( retry < 0 ) {
+			throw new SearchException( Environment.RETRY_MARKER_LOOKUP +
+					" option must be a positive integer, but was \"" + retry + "\"" );
 		}
 		boolean currentMarkerInSource = false;
 		for ( int tried = 0 ; tried <= retry ; tried++ ) {
@@ -119,17 +120,30 @@ private void checkCurrentMarkerInSource() {
 			}
 			currentMarkerInSource =
 					new File( sourceIndexDir, "current1" ).exists()
-					|| new File(sourceIndexDir, "current2").exists();
-			if (currentMarkerInSource) {
+					|| new File( sourceIndexDir, "current2" ).exists();
+			if ( currentMarkerInSource ) {
 				break;
 			}
 		}
-		if ( !currentMarkerInSource ) {
-			throw new IllegalStateException( "No current marker in source directory. Has the master being started once already?" );
-		}
+		return currentMarkerInSource;
 	}
 
 	public void start() {
+		if ( ! attemptInitializeAndStart() ) {
+			// if we failed to initialize and/or start, we'll try again later: setup a timer
+			long period = DirectoryProviderHelper.getRetryInitializePeriod( properties, directoryProviderName );
+			if ( period != 0 ) {
+				scheduleTask( new InitTask(), period );
+			}
+			else {
+				throw new SearchException( "Failed to initialize DirectoryProvider \""
+						+ directoryProviderName + "\": could not find marker file in index source" );
+			}
+		}
+	}
+
+	private void startIt() {
+		@SuppressWarnings("unused")
 		int readCurrentState = current; //Unneeded value, but ensure visibility of state protected by memory barrier
 		int currentToBe = 0;
 		try {
@@ -179,13 +193,23 @@ else if ( new File( sourceIndexDir, "current2" ).exists() ) {
 		catch ( IOException e ) {
 			throw new SearchException( "Unable to initialize index: " + directoryProviderName, e );
 		}
-		task = new TriggerTask( sourceIndexDir, indexDir );
+		updateTask = new UpdateTask( sourceIndexDir, indexDir );
 		long period = DirectoryProviderHelper.getRefreshPeriod( properties, directoryProviderName );
-		timer.scheduleAtFixedRate( task, period, period );
+		scheduleTask( updateTask, period );
 		this.current = currentToBe;
+		started = true;
 	}
 
-	public FSDirectory getDirectory() {
+	public Directory getDirectory() {
+		if ( !started ) {
+			if ( dummyDirectory == null ) {
+				RAMDirectory directory = new RAMDirectory();
+				DirectoryProviderHelper.initializeIndexIfNeeded( directory );
+				dummyDirectory = directory;
+			}
+			return dummyDirectory;
+		}
+
 		int readState = current;// to have the read consistent in the next two "if"s.
 		if ( readState == 1 ) {
 			return directory1;
@@ -223,17 +247,53 @@ public int hashCode() {
 		// but from a practical POV this is fine since we only call this method
 		// after initialize call
 		@SuppressWarnings("unused")
-		int readCurrentState = current; //unneded value, but ensure visibility of indexName
+		int readCurrentState = current; //unneeded value, but ensure visibility of indexName
 		int hash = 11;
 		return 37 * hash + indexName.hashCode();
 	}
 
-	class TriggerTask extends TimerTask {
+	class InitTask extends TimerTask {
+
+		@Override
+		public void run() {
+			try {
+				if ( attemptInitializeAndStart() ) {
+					// then this task is no longer needed
+					cancel();
+				}
+			}
+			catch (RuntimeException re) {
+				// we need this to make sure the error is logged somewhere,
+				// as we're executing it in the timer thread
+				log.error( "Failed to initialize SlaveDirectoryProvider " + indexName, re );
+			}
+		}
+	}
+
+	/**
+	 * @return true if both initialize and start succeeded
+	 */
+	protected synchronized boolean attemptInitializeAndStart() {
+		if ( !initialized ) {
+			if ( currentMarkerIsInSource() ) {
+				initialized = true;
+				log.info( "Found current marker in source directory - initialization succeeded" );
+			} else {
+				log.warn( "No current marker in source directory. Has the master being started already?" );
+			}
+		}
+		if ( initialized ) {
+			startIt();
+		}
+		return this.started;
+	}
+
+	class UpdateTask extends TimerTask {
 
 		private final ExecutorService executor;
 		private final CopyDirectory copyTask;
 
-		public TriggerTask(File sourceIndexDir, File destination) {
+		public UpdateTask(File sourceIndexDir, File destination) {
 			executor = Executors.newSingleThreadExecutor();
 			copyTask = new CopyDirectory( sourceIndexDir, destination );
 		}
@@ -338,20 +398,27 @@ else if ( new File( source, "current2" ).exists() ) {
 
 	public void stop() {
 		@SuppressWarnings("unused")
-		int readCurrentState = current; //unneded value, but ensure visibility of state protected by memory barrier
+		int readCurrentState = current; //unneeded value, but ensure visibility of state protected by memory barrier
 		timer.cancel();
-		task.stop();
-		try {
-			directory1.close();
-		}
-		catch ( Exception e ) {
-			log.error( "Unable to properly close Lucene directory {}" + directory1.getFile(), e );
-		}
-		try {
-			directory2.close();
+		if ( updateTask != null ) {
+			updateTask.stop();
 		}
-		catch ( Exception e ) {
-			log.error( "Unable to properly close Lucene directory {}" + directory2.getFile(), e );
+		closeDirectory( directory1 );
+		closeDirectory( directory2 );
+	}
+
+	private void closeDirectory(Directory directory) {
+		if ( directory != null ) {
+			try {
+				directory.close();
+			} catch ( Exception e ) {
+				log.error( "Unable to properly close Lucene directory " + directory, e );
+			}
 		}
 	}
+
+	protected void scheduleTask(TimerTask task, long period) {
+		timer.schedule( task, period, period );
+	}
+
 }

hibernate-search/src/test/java/org/hibernate/search/test/directoryProvider/FSSlaveAndMasterDPTest.java

@@ -52,7 +52,7 @@ public class FSSlaveAndMasterDPTest extends MultipleSFTestCase {
 
 	private static final Logger log = LoggerFactory.make();
 
-	private static File root;
+	static File root;
 
 	static {
 		String buildDir = System.getProperty( "build.dir" );
@@ -66,22 +66,22 @@ public class FSSlaveAndMasterDPTest extends MultipleSFTestCase {
 	/**
 	 * The lucene index directory which is shared between master and slave.
 	 */
-	private String masterCopy = "/master/copy";
+	final static String masterCopy = "/master/copy";
 
 	/**
 	 * The lucene index directory which is specific to the master node.
 	 */
-	private String masterMain = "/master/main";
+	final static String masterMain = "/master/main";
 
 	/**
 	 * The lucene index directory which is specific to the slave node.
 	 */
-	private String slave = "/slave";
+	final static String slave = "/slave";
 
 	/**
 	 * The lucene index directory which is specific to the slave node.
 	 */
-	private String slaveUnready = "/slaveUnready";
+	final static String slaveUnready = "/slaveUnready";
 
 	/**
 	 * Verifies that copies of the master get properly copied to the slaves.
@@ -180,9 +180,8 @@ private Session getMasterSession() {
 	private Session getSlaveSession() {
 		return getSessionFactories()[1].openSession();
 	}
-
-	protected void setUp() throws Exception {
-
+
+	static void prepareDirectories() {
 		if ( root.exists() ) {
 			FileHelper.delete( root );
 		}
@@ -210,11 +209,19 @@ protected void setUp() throws Exception {
 		if ( !slaveUnreadyFile.mkdirs() ) {
 			throw new HibernateException( "Unable to setup slave directory" );
 		}
+	}
+
+	protected void setUp() throws Exception {
+		prepareDirectories();
 		super.setUp();
 	}
 
 	protected void tearDown() throws Exception {
 		super.tearDown();
+		cleanupDirectories();
+	}
+
+	static void cleanupDirectories() {
 		log.info( "Deleting test directory {} ", root.getAbsolutePath() );
 		FileHelper.delete( root );
 	}
@@ -269,7 +276,7 @@ protected void configure(Configuration[] cfg) {
 		cfg[1].setProperty( "hibernate.search.default.refresh", "1" ); //every second
 		//keep the fqcn to make sure non short cut solutions still work
 		cfg[1].setProperty(
-				"hibernate.search.default.directory_provider", "org.hibernate.search.store.FSSlaveDirectoryProvider"
+				"hibernate.search.default.directory_provider", "filesystem-slave"
 		);
 	}
 }