Skip to content

Add handling for unexpectedly closed IndexWriter #55

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Apr 7, 2025
Merged

Add handling for unexpectedly closed IndexWriter #55

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c1f4ce2
Add handling for unexpectedly closed IndexWriter
patrick-austin Apr 4, 2025
68d4aed
Create new SearcherManager if IndexWriter is re-opened
patrick-austin Apr 7, 2025
2ca22b5
Increase IOException logging
patrick-austin Apr 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 56 additions & 11 deletions src/main/java/org/icatproject/lucene/Lucene.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,9 +104,9 @@ public class Lucene {
* A bucket for accessing the read and write functionality for a single "shard"
* Lucene index which can then be grouped to represent a single document type.
*/
private class ShardBucket {
public class ShardBucket {
private FSDirectory directory;
private IndexWriter indexWriter;
public IndexWriter indexWriter;
private SearcherManager searcherManager;
private DefaultSortedSetDocValuesReaderState state;
private AtomicLong documentCount;
Expand Down Expand Up @@ -147,8 +147,10 @@ public ShardBucket(java.nio.file.Path shardPath) throws IOException {
*
* @return The number of documents committed to this shard.
* @throws IOException
* @throws LuceneException If the IndexWriter is closed
*/
public int commit() throws IOException {
public int commit() throws IOException, LuceneException {
ensureOpen();
if (indexWriter.hasUncommittedChanges()) {
indexWriter.commit();
searcherManager.maybeRefreshBlocking();
Expand Down Expand Up @@ -183,6 +185,32 @@ private void initState(IndexSearcher indexSearcher) throws IOException {
searcherManager.release(indexSearcher);
}
}

/**
* To be called before attempting to commit. If the IndexWriter has
* been closed, e.g. due to a previous IOException, will create a new
* one and throw (so that whatever process called this knows the commit
* failed and changes will not have been persisted).
*
* @throws IOException If index directory cannot be read/written
* @throws LuceneException If the IndexWriter is closed
*/
public void ensureOpen() throws IOException, LuceneException {
if (!indexWriter.isOpen()) {
IndexWriterConfig config = new IndexWriterConfig(analyzer);
indexWriter = new IndexWriter(directory, config);
searcherManager = new SearcherManager(indexWriter, null);
IndexSearcher indexSearcher = searcherManager.acquire();
int numDocs = indexSearcher.getIndexReader().numDocs();
documentCount = new AtomicLong(numDocs);
initState(indexSearcher);

String fileName = directory.getDirectory().getFileName().toString();
String message = "IndexWriter for " + fileName + " was unexpectedly closed";
logger.error(message);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, message);
}
}
}

/**
Expand Down Expand Up @@ -245,8 +273,9 @@ public List<IndexSearcher> acquireSearchers() throws IOException {
*
* @param document The document to be added.
* @throws IOException
* @throws LuceneException If the IndexWriter is closed
*/
public void addDocument(Document document) throws IOException {
public void addDocument(Document document) throws IOException, LuceneException {
ShardBucket shardBucket = routeShard();
shardBucket.indexWriter.addDocument(document);
shardBucket.documentCount.incrementAndGet();
Expand Down Expand Up @@ -282,8 +311,9 @@ public void deleteDocuments(String field, long value) throws IOException {
* @param icatId The ICAT id of the document to be updated.
* @param document The document that will replace the old document.
* @throws IOException
* @throws LuceneException If the IndexWriter is closed
*/
public void updateDocument(long icatId, Document document) throws IOException {
public void updateDocument(long icatId, Document document) throws IOException, LuceneException {
deleteDocument(icatId);
addDocument(document);
}
Expand Down Expand Up @@ -311,8 +341,9 @@ public ShardBucket buildShardBucket(int shardKey) throws IOException {
* @param entityName The name of the entities being committed. Only used for
* debug logging.
* @throws IOException
* @throws LuceneException If the IndexWriter is closed
*/
public void commit(String command, String entityName) throws IOException {
public void commit(String command, String entityName) throws IOException, LuceneException {
for (ShardBucket shardBucket : shardList) {
int cached = shardBucket.commit();
if (cached != 0) {
Expand All @@ -332,8 +363,10 @@ public void commit(String command, String entityName) throws IOException {
public void close() throws IOException {
for (ShardBucket shardBucket : shardList) {
shardBucket.searcherManager.close();
shardBucket.indexWriter.commit();
shardBucket.indexWriter.close();
if (shardBucket.indexWriter.isOpen()) {
shardBucket.indexWriter.commit();
shardBucket.indexWriter.close();
}
shardBucket.directory.close();
}
}
Expand All @@ -354,10 +387,12 @@ public ShardBucket getCurrentShardBucket() {
*
* @return The ShardBucket that the relevant Document is/should be indexed in.
* @throws IOException
* @throws LuceneException If the IndexWriter is closed
*/
public ShardBucket routeShard() throws IOException {
public ShardBucket routeShard() throws IOException, LuceneException {
ShardBucket shardBucket = getCurrentShardBucket();
if (shardBucket.documentCount.get() >= luceneMaxShardSize) {
shardBucket.ensureOpen();
shardBucket.indexWriter.commit();
shardBucket = buildShardBucket(shardList.size());
}
Expand Down Expand Up @@ -446,6 +481,7 @@ public void modify(@Context HttpServletRequest request) throws LuceneException {
}
count = operations.size();
} catch (IOException e) {
logger.error("IOException while attempting to modify document(s)", e);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, e.getMessage());
}
logger.debug("Modified {} documents", count);
Expand Down Expand Up @@ -474,6 +510,7 @@ public void addNow(@Context HttpServletRequest request, @PathParam("entityName")
logger.error("Could not parse JSON from {}", value);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, e.getMessage());
} catch (IOException e) {
logger.error("IOException while attempting to add document(s)", e);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, e.getMessage());
}
logger.debug("Added {} {} documents", documents.size(), entityName);
Expand All @@ -495,6 +532,7 @@ public void clear() throws LuceneException {
Files.walk(luceneDirectory, FileVisitOption.FOLLOW_LINKS).sorted(Comparator.reverseOrder())
.filter(f -> !luceneDirectory.equals(f)).map(java.nio.file.Path::toFile).forEach(File::delete);
} catch (IOException e) {
logger.error("IOException while attempting to clear", e);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, e.getMessage());
}

Expand All @@ -519,6 +557,7 @@ public void commit() throws LuceneException {
}
}
} catch (IOException e) {
logger.error("IOException while attempting to commit", e);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, e.getMessage());
}
}
Expand Down Expand Up @@ -571,9 +610,10 @@ private void create(JsonObject operationBody) throws NumberFormatException, IOEx
* @param entityId Icat id of entity to update as a JsonNumber.
* @param index Index (entity) to update.
* @throws IOException
* @throws LuceneException If the IndexWriter is closed
*/
private void aggregateFileSize(long sizeToAdd, long sizeToSubtract, long deltaFileCount, JsonNumber entityId,
String index) throws IOException {
String index) throws IOException, LuceneException {
if (entityId != null) {
aggregateFileSize(sizeToAdd, sizeToSubtract, deltaFileCount, entityId.longValueExact(), index);
}
Expand All @@ -592,9 +632,10 @@ private void aggregateFileSize(long sizeToAdd, long sizeToSubtract, long deltaFi
* @param entityId Icat id of entity to update as a long.
* @param index Index (entity) to update.
* @throws IOException
* @throws LuceneException If the IndexWriter is closed
*/
private void aggregateFileSize(long sizeToAdd, long sizeToSubtract, long deltaFileCount, long entityId,
String index) throws IOException {
String index) throws IOException, LuceneException {
long deltaFileSize = sizeToAdd - sizeToSubtract;
if (deltaFileSize != 0 || deltaFileCount != 0) {
IndexBucket indexBucket = indexBuckets.computeIfAbsent(index, k -> new IndexBucket(k));
Expand Down Expand Up @@ -766,6 +807,7 @@ private void delete(JsonObject operationBody) throws LuceneException, IOExceptio
shardBucket.indexWriter.deleteDocuments(idQuery);
}
} catch (IOException e) {
logger.error("IOException while attempting to delete document(s)", e);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, e.getMessage());
}
}
Expand Down Expand Up @@ -921,6 +963,7 @@ public void freeSearcher(SearchBucket search) throws LuceneException {
indexBuckets.computeIfAbsent(name.toLowerCase(), k -> new IndexBucket(k))
.releaseSearchers(subReaders);
} catch (IOException e) {
logger.error("IOException while attempting to freeSearcher", e);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, e.getMessage());
}
}
Expand Down Expand Up @@ -1136,6 +1179,7 @@ public void lock(@PathParam("entityName") String entityName, @QueryParam("minId"
}
}
} catch (IOException e) {
logger.error("IOException while attempting to lock", e);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, e.getMessage());
}
}
Expand Down Expand Up @@ -1699,6 +1743,7 @@ public void unlock(@PathParam("entityName") String entityName) throws LuceneExce
try {
bucket.commit("Unlock", entityName);
} catch (IOException e) {
logger.error("IOException while attempting to unlock", e);
throw new LuceneException(HttpURLConnection.HTTP_INTERNAL_ERROR, e.getMessage());
}
}
Expand Down
16 changes: 16 additions & 0 deletions src/test/java/icat/lucene/TestLucene.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;

import java.io.IOException;
Expand Down Expand Up @@ -52,6 +54,7 @@
import org.icatproject.lucene.IcatSynonymAnalyzer;
import org.icatproject.lucene.Lucene;
import org.icatproject.lucene.SearchBucket;
import org.icatproject.lucene.Lucene.ShardBucket;
import org.icatproject.lucene.SearchBucket.SearchType;
import org.icatproject.lucene.exceptions.LuceneException;
import org.junit.Test;
Expand All @@ -65,6 +68,19 @@ public class TestLucene {

private final FacetsConfig facetsConfig = new FacetsConfig();

@Test
public void testEnsureOpen() throws Exception {
Lucene lucene = new Lucene();
Path tmpLuceneDir = Files.createTempDirectory("lucene");
ShardBucket shardBucket = lucene.new ShardBucket(tmpLuceneDir.resolve("Investigation"));
shardBucket.ensureOpen(); // Should not throw as still open
shardBucket.indexWriter.close();
IndexWriter closedIndexWriter = shardBucket.indexWriter;
assertThrows(LuceneException.class, () -> shardBucket.ensureOpen());
assertNotSame(closedIndexWriter, shardBucket.indexWriter);
assertTrue(shardBucket.indexWriter.isOpen());
}

@Test
public void testIcatAnalyzer() throws Exception {
final String text = "This is a demo of the 1st (or is it number 2) all singing and dancing TokenStream's API with added aardvarks";
Expand Down