Initial working implementation with new tests. Needs more testing and polish

Author: nfsantos

oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/ElasticIndexerProvider.java

@@ -24,6 +24,7 @@
 import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexDefinition;
 import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexTracker;
 import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticMetricHandler;
+import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticBulkProcessorHandler;
 import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticDocument;
 import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticIndexEditorProvider;
 import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticIndexWriterFactory;
@@ -39,6 +40,7 @@
 
 import org.jetbrains.annotations.Nullable;
 
+import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
 import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.TYPE_PROPERTY_NAME;
@@ -49,12 +51,15 @@ public class ElasticIndexerProvider implements NodeStateIndexerProvider {
     private final IndexHelper indexHelper;
     private final ElasticIndexWriterFactory indexWriterFactory;
     private final ElasticConnection connection;
+    private final ElasticBulkProcessorHandler bulkProcessorHandler;
 
     public ElasticIndexerProvider(IndexHelper indexHelper, ElasticConnection connection) {
         this.indexHelper = indexHelper;
-        this.indexWriterFactory = new ElasticIndexWriterFactory(connection,
-                new ElasticIndexTracker(connection, new ElasticMetricHandler(StatisticsProvider.NOOP)));
         this.connection = connection;
+        this.bulkProcessorHandler = new ElasticBulkProcessorHandler(connection);
+        this.indexWriterFactory = new ElasticIndexWriterFactory(connection,
+                new ElasticIndexTracker(connection, new ElasticMetricHandler(StatisticsProvider.NOOP)), bulkProcessorHandler);
+
     }
 
     @Override
@@ -79,5 +84,11 @@ public ExtractedTextCache getTextCache() {
     }
 
     @Override
-    public void close() {}
+    public void close() {
+        try {
+            this.bulkProcessorHandler.close();
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }

oak-run-elastic/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/ElasticIndexerTest.java

@@ -26,6 +26,8 @@
 import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticConnection;
 import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexDefinition;
 import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexTracker;
+import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticBulkProcessorHandler;
+import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticDocument;
 import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticIndexEditorProvider;
 import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticIndexWriterFactory;
 import org.apache.jackrabbit.oak.plugins.index.elastic.util.ElasticIndexDefinitionBuilder;
@@ -71,8 +73,8 @@ public void nodeIndexed_WithIncludedPaths() throws Exception {
         when(elasticsearchAsyncClientMock._jsonpMapper()).thenReturn(jsonMapperMock);
         when(elasticConnectionMock.getAsyncClient()).thenReturn(elasticsearchAsyncClientMock);
 
-        FulltextIndexWriter indexWriter = new ElasticIndexWriterFactory(elasticConnectionMock,
-                mock(ElasticIndexTracker.class)).newInstance(idxDefn, defn.builder(), CommitInfo.EMPTY, false);
+        FulltextIndexWriter<ElasticDocument> indexWriter = new ElasticIndexWriterFactory(elasticConnectionMock,
+                mock(ElasticIndexTracker.class), mock(ElasticBulkProcessorHandler.class)).newInstance(idxDefn, defn.builder(), CommitInfo.EMPTY, false);
         ElasticIndexer indexer = new ElasticIndexer(idxDefn, mock(FulltextBinaryTextExtractor.class), builder,
                 mock(IndexingProgressReporter.class), indexWriter, mock(ElasticIndexEditorProvider.class), mock(IndexHelper.class));
 

oak-search-elastic/src/main/java/org/apache/jackrabbit/oak/plugins/index/elastic/ElasticIndexDefinition.java

@@ -44,15 +44,6 @@ public class ElasticIndexDefinition extends IndexDefinition {
 
     public static final String TYPE_ELASTICSEARCH = "elasticsearch";
 
-    public static final String BULK_ACTIONS = "bulkActions";
-    public static final int BULK_ACTIONS_DEFAULT = 250;
-
-    public static final String BULK_SIZE_BYTES = "bulkSizeBytes";
-    public static final long BULK_SIZE_BYTES_DEFAULT = 1024 * 1024; // 1MB
-
-    public static final String BULK_FLUSH_INTERVAL_MS = "bulkFlushIntervalMs";
-    public static final long BULK_FLUSH_INTERVAL_MS_DEFAULT = 3000;
-
     public static final String NUMBER_OF_SHARDS = "numberOfShards";
     public static final int NUMBER_OF_SHARDS_DEFAULT = 1;
 
@@ -82,10 +73,6 @@ public class ElasticIndexDefinition extends IndexDefinition {
     public static final String LIMIT_TOTAL_FIELDS = "limitTotalFields";
     public static final long LIMIT_TOTAL_FIELDS_DEFAULT = 1000L;
 
-    // when true, fails indexing in case of bulk failures
-    public static final String FAIL_ON_ERROR = "failOnError";
-    public static final boolean FAIL_ON_ERROR_DEFAULT = true;
-
     /**
      * When 0, the index name gets dynamically generated by adding a random suffix to the index name.
      */
@@ -178,9 +165,6 @@ public class ElasticIndexDefinition extends IndexDefinition {
 
     private final String indexPrefix;
     private final String indexAlias;
-    public final int bulkActions;
-    public final long bulkSizeBytes;
-    public final long bulkFlushIntervalMs;
     private final boolean similarityTagsEnabled;
     private final float similarityTagsBoost;
     public final int numberOfShards;
@@ -189,7 +173,6 @@ public class ElasticIndexDefinition extends IndexDefinition {
     public final long queryTimeoutMs;
     public final Integer trackTotalHits;
     public final String dynamicMapping;
-    public final boolean failOnError;
     public final long indexNameSeed;
     public final InferenceDefinition inferenceDefinition;
     public final long limitTotalFields;
@@ -211,9 +194,6 @@ public ElasticIndexDefinition(NodeState root, NodeState defn, String indexPath,
         } else {
             this.indexAlias = ElasticIndexNameHelper.getElasticSafeIndexName(indexPrefix, getIndexPath());
         }
-        this.bulkActions = getOptionalValue(defn, BULK_ACTIONS, BULK_ACTIONS_DEFAULT);
-        this.bulkSizeBytes = getOptionalValue(defn, BULK_SIZE_BYTES, BULK_SIZE_BYTES_DEFAULT);
-        this.bulkFlushIntervalMs = getOptionalValue(defn, BULK_FLUSH_INTERVAL_MS, BULK_FLUSH_INTERVAL_MS_DEFAULT);
         this.numberOfShards = getOptionalValue(defn, NUMBER_OF_SHARDS, NUMBER_OF_SHARDS_DEFAULT);
         this.numberOfReplicas = getOptionalValue(defn, NUMBER_OF_REPLICAS, NUMBER_OF_REPLICAS_DEFAULT);
         this.similarityTagsEnabled = getOptionalValue(defn, SIMILARITY_TAGS_ENABLED, SIMILARITY_TAGS_ENABLED_DEFAULT);
@@ -223,9 +203,6 @@ public ElasticIndexDefinition(NodeState root, NodeState defn, String indexPath,
         this.queryTimeoutMs = getOptionalValue(defn, QUERY_TIMEOUT_MS, QUERY_TIMEOUT_MS_DEFAULT);
         this.trackTotalHits = getOptionalValue(defn, TRACK_TOTAL_HITS, TRACK_TOTAL_HITS_DEFAULT);
         this.dynamicMapping = getOptionalValue(defn, DYNAMIC_MAPPING, DYNAMIC_MAPPING_DEFAULT);
-        this.failOnError = getOptionalValue(defn, FAIL_ON_ERROR,
-                Boolean.parseBoolean(System.getProperty(TYPE_ELASTICSEARCH + "." + FAIL_ON_ERROR, Boolean.toString(FAIL_ON_ERROR_DEFAULT)))
-        );
         this.indexNameSeed = getOptionalValue(defn, INDEX_NAME_SEED, INDEX_NAME_SEED_DEFAULT);
         this.similarityTagsFields = getOptionalValues(defn, SIMILARITY_TAGS_FIELDS, Type.STRINGS, String.class, SIMILARITY_TAGS_FIELDS_DEFAULT);
         this.limitTotalFields = getOptionalValue(defn, LIMIT_TOTAL_FIELDS, LIMIT_TOTAL_FIELDS_DEFAULT);