Remove synchronized blocks from FDBDirectoryWrapper (#3494)

ForkJoinPool doesn't play well with `synchronized` blocks, particularly,
it can result in thread starvation / deadlocks. Threads waiting on
`synchronized` blocks, still count as active, and thus count towards the
parallelism of the pool.
The tests here run in a very small pool, to make sure that this won't
happen, but in actual production workloads, one would expect a much
larger pool, but also a lot more concurrent activity.

These tests were failing pretty reliably before with a pool size of 3,
and now they seem to pass reliably.

One change in here that may be surprising is the change around the
`writerReader`. I looked at the usages, and there were two:
1. Cursors would call `getReader` which would always flush, and thus
never use the field
2. `LuceneIndexMaintainer.deleteDocument` would never flush

So it makes sense to have these be two independent calls, which allows
`getWriterReader` to use a `LazyCloseable`.

Resolves: #2989
Resolves: #3146

Author: ScottDugas

fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintainer.java

@@ -299,7 +299,7 @@ int deleteDocument(Tuple groupingKey, Integer partitionId, Tuple primaryKey) thr
         @Nullable final LucenePrimaryKeySegmentIndex segmentIndex = directoryManager.getDirectory(groupingKey, partitionId).getPrimaryKeySegmentIndex();
 
         if (segmentIndex != null) {
-            final DirectoryReader directoryReader = directoryManager.getDirectoryReader(groupingKey, partitionId);
+            final DirectoryReader directoryReader = directoryManager.getWriterReader(groupingKey, partitionId);
             final LucenePrimaryKeySegmentIndex.DocumentIndexEntry documentIndexEntry = segmentIndex.findDocument(directoryReader, primaryKey);
             if (documentIndexEntry != null) {
                 state.context.ensureActive().clear(documentIndexEntry.entryKey); // TODO: Only if valid?
@@ -702,14 +702,15 @@ private CompletableFuture<IndexOperationResult> getLuceneInfoForAllPartitions(fi
     @SuppressWarnings("PMD.CloseResource") // the indexReader and directory are closed by the directoryManager
     private CompletableFuture<LuceneMetadataInfo.LuceneInfo> getLuceneInfo(final Tuple groupingKey, final Integer partitionId) {
         try {
-            final IndexReader indexReader = directoryManager.getIndexReader(groupingKey, partitionId);
-            final FDBDirectory directory = getDirectory(groupingKey, partitionId);
-            final CompletableFuture<Integer> fieldInfosFuture = directory.getFieldInfosCount();
-            return directory.listAllAsync()
-                    .thenCombine(fieldInfosFuture, (fileList, fieldInfosCount) ->
-                            new LuceneMetadataInfo.LuceneInfo(
-                                    indexReader.numDocs(),
-                                    fileList, fieldInfosCount));
+            try (IndexReader indexReader = directoryManager.getIndexReader(groupingKey, partitionId)) {
+                final FDBDirectory directory = getDirectory(groupingKey, partitionId);
+                final CompletableFuture<Integer> fieldInfosFuture = directory.getFieldInfosCount();
+                return directory.listAllAsync()
+                        .thenCombine(fieldInfosFuture, (fileList, fieldInfosCount) ->
+                                new LuceneMetadataInfo.LuceneInfo(
+                                        indexReader.numDocs(),
+                                        fileList, fieldInfosCount));
+            }
         } catch (IOException e) {
             return CompletableFuture.failedFuture(e);
         }

fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneIndexScrubbingToolsMissing.java

@@ -198,7 +198,7 @@ private boolean isMissingIndexKey(FDBIndexableRecord<Message> rec, Integer parti
             throw LuceneExceptions.toRecordCoreException("failed getIndexWriter", e);
         }
         try {
-            DirectoryReader directoryReader = directoryManager.getDirectoryReader(groupingKey, partitionId);
+            DirectoryReader directoryReader = directoryManager.getWriterReader(groupingKey, partitionId);
             final LucenePrimaryKeySegmentIndex.DocumentIndexEntry documentIndexEntry = segmentIndex.findDocument(directoryReader, rec.getPrimaryKey());
             if (documentIndexEntry == null) {
                 // Here: the document had not been found in the PK segment index

fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryManager.java

@@ -84,22 +84,18 @@ public class FDBDirectoryManager implements AutoCloseable {
     @Nonnull
     private final Map<Tuple, FDBDirectoryWrapper> createdDirectories;
     private final int mergeDirectoryCount;
-    @Nullable
-    private final Exception exceptionAtCreation;
     @Nonnull
     protected final LuceneAnalyzerWrapper writerAnalyzer;
     @Nonnull
     private final LuceneAnalyzerCombinationProvider analyzerSelector;
+    @Nullable
+    protected final Exception exceptionAtCreation;
 
     protected FDBDirectoryManager(@Nonnull IndexMaintainerState state) {
         this.state = state;
         this.createdDirectories = new ConcurrentHashMap<>();
         this.mergeDirectoryCount = getMergeDirectoryCount(state);
-        if (FDBTieredMergePolicy.usesCreationStack()) {
-            this.exceptionAtCreation = new Exception();
-        } else {
-            this.exceptionAtCreation = null;
-        }
+        this.exceptionAtCreation = FDBTieredMergePolicy.usesCreationStack() ? new Exception() : null;
         final var fieldInfos = LuceneIndexExpressions.getDocumentFieldDerivations(state.index, state.store.getRecordMetaData());
         this.analyzerSelector = LuceneAnalyzerRegistryImpl.instance().getLuceneAnalyzerCombinationProvider(state.index, LuceneAnalyzerType.FULL_TEXT, fieldInfos);
         this.writerAnalyzer = analyzerSelector.provideIndexAnalyzer();
@@ -206,7 +202,7 @@ public void mergeIndexWithContext(@Nonnull final Tuple groupingKey,
                                       @Nonnull final AgilityContext agilityContext) {
         try (FDBDirectoryWrapper directoryWrapper = createDirectoryWrapper(groupingKey, partitionId, agilityContext)) {
             try {
-                directoryWrapper.mergeIndex(exceptionAtCreation);
+                directoryWrapper.mergeIndex();
                 if (LOGGER.isDebugEnabled()) {
                     LOGGER.debug(KeyValueLogMessage.of("Lucene merge success",
                             LuceneLogMessageKeys.GROUP, groupingKey,
@@ -312,7 +308,8 @@ private FDBDirectoryWrapper createDirectoryWrapper(@Nullable Tuple groupingKey,
     }
 
     protected @Nonnull FDBDirectoryWrapper createNewDirectoryWrapper(final IndexMaintainerState state, final Tuple key, final int mergeDirectoryCount, final AgilityContext agilityContext, final int blockCacheMaximumSize) {
-        return new FDBDirectoryWrapper(state, key, mergeDirectoryCount, agilityContext, blockCacheMaximumSize, writerAnalyzer);
+        return new FDBDirectoryWrapper(state, key, mergeDirectoryCount, agilityContext, blockCacheMaximumSize,
+                writerAnalyzer, exceptionAtCreation);
     }
 
     private int getBlockCacheMaximumSize() {
@@ -367,11 +364,11 @@ public IndexReader getIndexReader(@Nullable Tuple groupingKey, @Nullable Integer
 
     @Nonnull
     public IndexWriter getIndexWriter(@Nullable Tuple groupingKey, @Nullable Integer partitionId) throws IOException {
-        return getDirectoryWrapper(groupingKey, partitionId).getWriter(exceptionAtCreation);
+        return getDirectoryWrapper(groupingKey, partitionId).getWriter();
     }
 
-    public DirectoryReader getDirectoryReader(@Nullable Tuple groupingKey, @Nullable Integer partititonId) throws IOException {
-        return getDirectoryWrapper(groupingKey, partititonId).getWriterReader(false);
+    public DirectoryReader getWriterReader(@Nullable Tuple groupingKey, @Nullable Integer partititonId) throws IOException {
+        return getDirectoryWrapper(groupingKey, partititonId).getWriterReader();
     }
 
     @Nonnull