Elasticsearch: Async handling of indexing/deletion requests

core/src/main/java/org/fao/geonet/kernel/DataManager.java

@@ -55,6 +55,7 @@
 import org.fao.geonet.kernel.datamanager.IMetadataValidator;
 import org.fao.geonet.kernel.schema.MetadataSchema;
 import org.fao.geonet.kernel.search.IndexingMode;
+import org.fao.geonet.kernel.search.submission.IIndexSubmitter;
 import org.fao.geonet.repository.UserGroupRepository;
 import org.jdom.Element;
 import org.slf4j.Logger;
@@ -65,7 +66,6 @@
 import javax.annotation.CheckForNull;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
-import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Collection;
 import java.util.List;
@@ -153,19 +153,14 @@ public void batchIndexInThreadPool(ServiceContext context, List<?> metadataIds)
         metadataIndexer.batchIndexInThreadPool(context, metadataIds);
     }
 
-    @Deprecated
-    public boolean isIndexing() {
-        return metadataIndexer.isIndexing();
-    }
-
     @Deprecated
     public void indexMetadata(final List<String> metadataIds) throws Exception {
         metadataIndexer.indexMetadata(metadataIds);
     }
 
     @Deprecated
-    public void indexMetadata(final String metadataId, boolean forceRefreshReaders) throws Exception {
-        metadataIndexer.indexMetadata(metadataId, forceRefreshReaders, IndexingMode.full);
+    public void indexMetadata(final String metadataId, IIndexSubmitter indexSubmittor) throws Exception {
+        metadataIndexer.indexMetadata(metadataId, indexSubmittor, IndexingMode.full);
     }
 
     @Deprecated
@@ -361,10 +356,10 @@ public String insertMetadata(ServiceContext context, String schema, Element meta
 
     @Deprecated
     public AbstractMetadata insertMetadata(ServiceContext context, AbstractMetadata newMetadata, Element metadataXml, IndexingMode indexingMode,
-                                           boolean updateFixedInfo, UpdateDatestamp updateDatestamp, boolean fullRightsForGroup, boolean forceRefreshReaders)
+                                           boolean updateFixedInfo, UpdateDatestamp updateDatestamp, boolean fullRightsForGroup, IIndexSubmitter indexSubmittor)
         throws Exception {
         return metadataManager.insertMetadata(context, newMetadata, metadataXml, indexingMode, updateFixedInfo, updateDatestamp,
-            fullRightsForGroup, forceRefreshReaders);
+            fullRightsForGroup, indexSubmittor);
     }
 
     @Deprecated
@@ -578,11 +573,6 @@ public void flush() {
         metadataManager.flush();
     }
 
-    @Deprecated
-    public void forceIndexChanges() throws IOException {
-        metadataIndexer.forceIndexChanges();
-    }
-
     @Deprecated
     public int batchDeleteMetadataAndUpdateIndex(Specification<? extends AbstractMetadata> specification) throws Exception {
         return metadataIndexer.batchDeleteMetadataAndUpdateIndex(specification);

core/src/main/java/org/fao/geonet/kernel/IndexMetadataTask.java

@@ -28,14 +28,12 @@
 import org.fao.geonet.Util;
 import org.fao.geonet.constants.Geonet;
 import org.fao.geonet.domain.User;
-import org.fao.geonet.kernel.search.EsSearchManager;
+import org.fao.geonet.kernel.search.submission.batch.BatchingIndexSubmitter;
 import org.fao.geonet.utils.Log;
 import org.springframework.transaction.TransactionStatus;
 
-import java.io.IOException;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -49,8 +47,6 @@ public final class IndexMetadataTask implements Runnable {
     private final List<?> _metadataIds;
     private final TransactionStatus _transactionStatus;
     private final Set<IndexMetadataTask> _batchIndex;
-    private final EsSearchManager searchManager;
-    private final AtomicInteger indexed;
     private User _user;
 
     /**
@@ -61,13 +57,11 @@ public final class IndexMetadataTask implements Runnable {
      * @param transactionStatus if non-null, wait for the transaction to complete before indexing
      */
     public IndexMetadataTask(@Nonnull ServiceContext context, @Nonnull List<?> metadataIds, Set<IndexMetadataTask> batchIndex,
-                      @Nullable TransactionStatus transactionStatus, @Nonnull AtomicInteger indexed) {
-        this.indexed = indexed;
+                      @Nullable TransactionStatus transactionStatus) {
         this._transactionStatus = transactionStatus;
         this._context = context;
         this._metadataIds = metadataIds;
         this._batchIndex = batchIndex;
-        this.searchManager = context.getBean(EsSearchManager.class);
 
         batchIndex.add(this);
 
@@ -77,7 +71,7 @@ public IndexMetadataTask(@Nonnull ServiceContext context, @Nonnull List<?> metad
     }
 
     public void run() {
-        try {
+        try (BatchingIndexSubmitter batchingIndexSubmittor = new BatchingIndexSubmitter(_metadataIds.size())) {
             _context.setAsThreadLocal();
             while (_transactionStatus != null && !_transactionStatus.isCompleted()) {
                 try {
@@ -101,13 +95,9 @@ public void run() {
             DataManager dataManager = _context.getBean(DataManager.class);
             // servlet up so safe to index all metadata that needs indexing
             for (Object metadataId : _metadataIds) {
-                this.indexed.incrementAndGet();
-                if (this.indexed.compareAndSet(500, 0)) {
-                    searchManager.forceIndexChanges();
-                }
 
                 try {
-                    dataManager.indexMetadata(metadataId.toString(), false);
+                    dataManager.indexMetadata(metadataId.toString(), batchingIndexSubmittor);
                 } catch (Exception e) {
                     Log.error(Geonet.INDEX_ENGINE, "Error indexing metadata '" + metadataId + "': " + e.getMessage()
                         + "\n" + Util.getStackTrace(e));
@@ -116,7 +106,6 @@ public void run() {
             if (_user != null && _context.getUserSession().getUserId() == null) {
                 _context.getUserSession().loginAs(_user);
             }
-            searchManager.forceIndexChanges();
         } finally {
             _batchIndex.remove(this);
         }

core/src/main/java/org/fao/geonet/kernel/datamanager/IMetadataIndexer.java

@@ -23,11 +23,11 @@
 
 package org.fao.geonet.kernel.datamanager;
 
-import java.io.IOException;
 import java.util.List;
 
 import org.fao.geonet.domain.AbstractMetadata;
 import org.fao.geonet.kernel.search.IndexingMode;
+import org.fao.geonet.kernel.search.submission.IIndexSubmitter;
 import org.jdom.Element;
 import org.springframework.data.jpa.domain.Specification;
 
@@ -49,11 +49,6 @@ public interface IMetadataIndexer {
      */
     public void init(ServiceContext context, Boolean force) throws Exception;
 
-    /**
-     * Force the index to wait until all changes are processed and the next reader obtained will get the latest data.
-     */
-    void forceIndexChanges() throws IOException;
-
     /**
      * Remove the records that matches the specification
      *
@@ -77,13 +72,6 @@ public interface IMetadataIndexer {
      */
     void batchIndexInThreadPool(ServiceContext context, List<?> metadataIds);
 
-    /**
-     * Is the platform currently indexing?
-     *
-     * @return
-     */
-    boolean isIndexing();
-
     /**
      * Index the list of records passed as parameter in order.
      *
@@ -92,7 +80,7 @@ public interface IMetadataIndexer {
      */
     void indexMetadata(List<String> metadataIds) throws Exception;
 
-    void indexMetadata(String metadataId, boolean forceRefreshReaders, IndexingMode indexingMode) throws Exception;
+    void indexMetadata(String metadataId, IIndexSubmitter indexSubmittor, IndexingMode indexingMode) throws Exception;
 
     void indexMetadataPrivileges(String uuid, int id) throws Exception;
 

core/src/main/java/org/fao/geonet/kernel/datamanager/IMetadataManager.java

@@ -33,6 +33,8 @@
 import org.fao.geonet.kernel.EditLib;
 import org.fao.geonet.kernel.UpdateDatestamp;
 import org.fao.geonet.kernel.search.IndexingMode;
+import org.fao.geonet.kernel.search.submission.IDeletionSubmitter;
+import org.fao.geonet.kernel.search.submission.IIndexSubmitter;
 import org.fao.geonet.repository.BatchUpdateQuery;
 import org.fao.geonet.repository.PathSpec;
 import org.fao.geonet.repository.Updater;
@@ -68,9 +70,10 @@ public interface IMetadataManager {
      *
      * @param context
      * @param metadataId
+     * @param deletionSubmittor
      * @throws Exception
      */
-    void deleteMetadata(ServiceContext context, String metadataId) throws Exception;
+    void deleteMetadata(ServiceContext context, String metadataId, IDeletionSubmitter deletionSubmittor) throws Exception;
 
     /**
      * Delete the record with the id metadataId
@@ -147,12 +150,12 @@ String insertMetadata(ServiceContext context, String schema, Element metadataXml
      * @param updateFixedInfo
      * @param updateDatestamp
      * @param fullRightsForGroup
-     * @param forceRefreshReaders
+     * @param indexSubmittor
      * @return
      * @throws Exception
      */
     AbstractMetadata insertMetadata(ServiceContext context, AbstractMetadata newMetadata, Element metadataXml, IndexingMode indexingMode,
-                                    boolean updateFixedInfo, UpdateDatestamp updateDatestamp, boolean fullRightsForGroup, boolean forceRefreshReaders)
+                                    boolean updateFixedInfo, UpdateDatestamp updateDatestamp, boolean fullRightsForGroup, IIndexSubmitter indexSubmittor)
             throws Exception;
 
     /**

core/src/main/java/org/fao/geonet/kernel/datamanager/base/BaseMetadataIndexer.java

@@ -49,6 +49,9 @@
 import org.fao.geonet.kernel.search.EsSearchManager;
 import org.fao.geonet.kernel.search.IndexFields;
 import org.fao.geonet.kernel.search.IndexingMode;
+import org.fao.geonet.kernel.search.submission.batch.BatchingDeletionSubmitter;
+import org.fao.geonet.kernel.search.submission.batch.BatchingIndexSubmitter;
+import org.fao.geonet.kernel.search.submission.IIndexSubmitter;
 import org.fao.geonet.kernel.setting.SettingManager;
 import org.fao.geonet.kernel.setting.Settings;
 import org.fao.geonet.repository.*;
@@ -69,13 +72,11 @@
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.interceptor.TransactionAspectSupport;
 
-import java.io.IOException;
 import java.nio.file.Path;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.fao.geonet.resources.Resources.DEFAULT_LOGO_EXTENSION;
 
@@ -147,11 +148,6 @@ public void setMetadataManager(IMetadataManager metadataManager) {
     Set<String> indexing = new HashSet<String>();
     Set<IndexMetadataTask> batchIndex = ConcurrentHashMap.newKeySet();
 
-    @Override
-    public void forceIndexChanges() throws IOException {
-        searchManager.forceIndexChanges();
-    }
-
     @Override
     public int batchDeleteMetadataAndUpdateIndex(Specification<? extends AbstractMetadata> specification)
         throws Exception {
@@ -173,29 +169,31 @@ public int batchDeleteMetadataAndUpdateIndex(Specification<? extends AbstractMet
 //        searchManager.delete(metadataToDelete.stream().map(input -> Integer.toString(input.getId())).collect(Collectors.toList()));
 //        metadataManager.deleteAll(specification);
         // So delete one by one even if slower
-        metadataToDelete.forEach(md -> {
-            try {
-                // Extract information for RecordDeletedEvent
-                LinkedHashMap<String, String> titles = metadataUtils.extractTitles(Integer.toString(md.getId()));
-                UserSession userSession = ServiceContext.get().getUserSession();
-                String xmlBefore = md.getData();
-
-                store.delResources(ServiceContext.get(), md.getUuid());
-                metadataManager.deleteMetadata(ServiceContext.get(), String.valueOf(md.getId()));
-
-                // Trigger RecordDeletedEvent
-                new RecordDeletedEvent(md.getId(), md.getUuid(), titles, userSession.getUserIdAsInt(), xmlBefore).publish(ApplicationContextHolder.get());
-            } catch (Exception e) {
-                Log.warning(Geonet.DATA_MANAGER, String.format(
-
-                    "Error during removal of metadata %s part of batch delete operation. " +
-                        "This error may create a ghost record (ie. not in the index " +
-                        "but still present in the database). " +
-                        "You can reindex the catalogue to see it again. " +
-                        "Error was: %s.", md.getUuid(), e.getMessage()));
-                e.printStackTrace();
-            }
-        });
+        try (BatchingDeletionSubmitter submitter = new BatchingDeletionSubmitter(metadataToDelete.size())) {
+            metadataToDelete.forEach(md -> {
+                try {
+                    // Extract information for RecordDeletedEvent
+                    LinkedHashMap<String, String> titles = metadataUtils.extractTitles(Integer.toString(md.getId()));
+                    UserSession userSession = ServiceContext.get().getUserSession();
+                    String xmlBefore = md.getData();
+
+                    store.delResources(ServiceContext.get(), md.getUuid());
+                    metadataManager.deleteMetadata(ServiceContext.get(), String.valueOf(md.getId()), submitter);
+
+                    // Trigger RecordDeletedEvent
+                    new RecordDeletedEvent(md.getId(), md.getUuid(), titles, userSession.getUserIdAsInt(), xmlBefore).publish(ApplicationContextHolder.get());
+                } catch (Exception e) {
+                    Log.warning(Geonet.DATA_MANAGER, String.format(
+
+                        "Error during removal of metadata %s part of batch delete operation. " +
+                            "This error may create a ghost record (ie. not in the index " +
+                            "but still present in the database). " +
+                            "You can reindex the catalogue to see it again. " +
+                            "Error was: %s.", md.getUuid(), e.getMessage()));
+                    e.printStackTrace();
+                }
+            });
+        }
 
         return metadataToDelete.size();
     }
@@ -268,7 +266,6 @@ public void batchIndexInThreadPool(ServiceContext context, List<?> metadataIds)
             Log.debug(Geonet.INDEX_ENGINE, "Indexing " + metadataIds.size() + " records.");
             Log.debug(Geonet.INDEX_ENGINE, metadataIds.toString());
         }
-        AtomicInteger numIndexedTracker = new AtomicInteger();
         while (index < metadataIds.size()) {
             int start = index;
             int count = Math.min(perThread, metadataIds.size() - start);
@@ -285,29 +282,26 @@ public void batchIndexInThreadPool(ServiceContext context, List<?> metadataIds)
             }
 
             // create threads to process this chunk of ids
-            Runnable worker = new IndexMetadataTask(context, subList, batchIndex, transactionStatus, numIndexedTracker);
+            Runnable worker = new IndexMetadataTask(context, subList, batchIndex, transactionStatus);
             executor.execute(worker);
             index += count;
         }
 
         executor.shutdown();
     }
 
-    @Override
-    public boolean isIndexing() {
-        return searchManager.isIndexing();
-    }
-
     @Override
     public void indexMetadata(final List<String> metadataIds) throws Exception {
-        for (String metadataId : metadataIds) {
-            indexMetadata(metadataId, true, IndexingMode.full);
+        try (BatchingIndexSubmitter indexSubmittor = new BatchingIndexSubmitter(metadataIds.size())) {
+            for (String metadataId : metadataIds) {
+                indexMetadata(metadataId, indexSubmittor, IndexingMode.full);
+            }
         }
     }
 
     @Override
     public void indexMetadata(final String metadataId,
-                              final boolean forceRefreshReaders,
+                              final IIndexSubmitter indexSubmittor,
                               final IndexingMode indexingMode)
         throws Exception {
         AbstractMetadata fullMd;
@@ -388,7 +382,7 @@ public void indexMetadata(final String metadataId,
                         "error",
                         Map.of("record", metadataId, "schema", schema)));
                 searchManager.index(null, md, indexKey, fields, metadataType,
-                    forceRefreshReaders, indexingMode);
+                    indexSubmittor, indexingMode);
                 Log.error(Geonet.DATA_MANAGER, String.format(
                     "Record %s / Schema '%s' is not registered in this catalog. Install it or remove those records. Record is indexed indexing error flag.",
                     metadataId, schema));
@@ -533,7 +527,7 @@ public void indexMetadata(final String metadataId,
                 }
 
                 searchManager.index(schemaManager.getSchemaDir(schema), md, indexKey, fields, metadataType,
-                    forceRefreshReaders, indexingMode);
+                    indexSubmittor, indexingMode);
             }
         } catch (Exception x) {
             Log.error(Geonet.DATA_MANAGER, "The metadata document index with id=" + metadataId