fix(index): Make bulk "index by query" operations configurable

commons/com.b2international.index/src/com/b2international/index/IndexClientFactory.java

@@ -73,6 +73,11 @@ public interface IndexClientFactory {
 	 */
 	String COMMIT_CONCURRENCY_LEVEL = "concurrencyLevel";
 
+	/**
+	 * Configuration key to specify the concurrency level for "update by query" and "delete by query" operations.
+	 */
+	String INDEX_BY_QUERY_CONCURRENCY_LEVEL = "indexByQueryConcurrencyLevel";
+
 	/**
 	 * Configuration key to specify the name of the embedded or TCP based Elasticsearch cluster to connect to.
 	 */
@@ -182,11 +187,17 @@ public interface IndexClientFactory {
 	int DEFAULT_MAX_TERMS_COUNT = 65_536;
 
 	/**
-	 * The default concurrency level for the bulk operations depends on the number of cores you have <code>max(1, cores / 4)</code>.
+	 * The default concurrency level for the bulk indexing operations depends on the number of cores you have <code>max(1, cores / 4)</code>.
 	 * Elasticsearch module only configuration key.
 	 */
 	int DEFAULT_COMMIT_CONCURRENCY_LEVEL = Math.max(1, Runtime.getRuntime().availableProcessors() / 4);
 
+	/**
+	 * The default maximum concurrency level for the "update by query" and "delete by query" operations is a constant value.
+	 * Elasticsearch module only configuration key.
+	 */
+	int DEFAULT_INDEX_BY_QUERY_CONCURRENCY_LEVEL = 4;
+
 	/**
 	 * The default index prefix is empty
 	 */

commons/com.b2international.index/src/com/b2international/index/es/EsDocumentWriter.java

@@ -120,11 +120,13 @@ public void commit() throws IOException {
 
 		final Set<DocumentMapping> mappingsToRefresh = Collections.synchronizedSet(newHashSet());
 		final EsClient client = admin.client();
+
 		// apply bulk updates first
 		final ListeningExecutorService executor;
 		admin.log().trace("Applying bulk updates ({}) and deletes ({})...", bulkUpdateOperations.size(), bulkDeleteOperations.size());
+
 		if (bulkUpdateOperations.size() > 1 || bulkDeleteOperations.size() > 1) {
-			final int threads = Math.min(4, Math.max(bulkUpdateOperations.size(), bulkDeleteOperations.size()));
+			final int threads = Math.min(getIndexByQueryConcurrencyLevel(), Math.max(bulkUpdateOperations.size(), bulkDeleteOperations.size()));
 			executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(threads));
 		} else {
 			executor = MoreExecutors.newDirectExecutorService();
@@ -261,6 +263,10 @@ public boolean isEmpty() {
 	private int getConcurrencyLevel() {
 		return (int) admin.settings().get(IndexClientFactory.COMMIT_CONCURRENCY_LEVEL);
 	}
+
+	private int getIndexByQueryConcurrencyLevel() {
+		return (int) admin.settings().get(IndexClientFactory.INDEX_BY_QUERY_CONCURRENCY_LEVEL);
+	}
 
 	/*
 	 * Testing only, dumps a text representation of all operations to the console

commons/com.b2international.index/src/com/b2international/index/es/admin/EsIndexAdmin.java

@@ -139,6 +139,7 @@ public EsIndexAdmin(EsClient client, ObjectMapper mapper, String name, Mappings
 
 		// local configuration settings for bulk writes, monitoring, etc.
 		this.settings.putIfAbsent(IndexClientFactory.COMMIT_CONCURRENCY_LEVEL, IndexClientFactory.DEFAULT_COMMIT_CONCURRENCY_LEVEL);
+		this.settings.putIfAbsent(IndexClientFactory.INDEX_BY_QUERY_CONCURRENCY_LEVEL, IndexClientFactory.DEFAULT_INDEX_BY_QUERY_CONCURRENCY_LEVEL);
 		this.settings.putIfAbsent(IndexClientFactory.BULK_ACTIONS_SIZE, IndexClientFactory.DEFAULT_BULK_ACTIONS_SIZE);
 		this.settings.putIfAbsent(IndexClientFactory.BULK_ACTIONS_SIZE_IN_MB, IndexClientFactory.DEFAULT_BULK_ACTIONS_SIZE_IN_MB);
 		this.settings.putIfAbsent(IndexClientFactory.COMMIT_WATERMARK_LOW_KEY, IndexClientFactory.DEFAULT_COMMIT_WATERMARK_LOW_VALUE);

core/com.b2international.snowowl.core/src/com/b2international/snowowl/core/config/IndexConfiguration.java

@@ -45,6 +45,8 @@ public class IndexConfiguration {
 	@Min(1)
 	private int commitConcurrencyLevel = Math.max(1, Runtime.getRuntime().availableProcessors() / 4);
 	@Min(1)
+	private int indexByQueryConcurrencyLevel = IndexClientFactory.DEFAULT_INDEX_BY_QUERY_CONCURRENCY_LEVEL;
+	@Min(1)
 	private int bulkActionSize = IndexClientFactory.DEFAULT_BULK_ACTIONS_SIZE;
 	@Min(1)
 	private int bulkActionSizeInMb = IndexClientFactory.DEFAULT_BULK_ACTIONS_SIZE_IN_MB;
@@ -102,6 +104,16 @@ public int getCommitConcurrencyLevel() {
 	public void setCommitConcurrencyLevel(int commitConcurrencyLevel) {
 		this.commitConcurrencyLevel = commitConcurrencyLevel;
 	}
+
+	@JsonProperty
+	public int getIndexByQueryConcurrencyLevel() {
+		return indexByQueryConcurrencyLevel;
+	}
+
+	@JsonProperty
+	public void setIndexByQueryConcurrencyLevel(int indexByQueryConcurrencyLevel) {
+		this.indexByQueryConcurrencyLevel = indexByQueryConcurrencyLevel;
+	}
 
 	@JsonProperty
 	public String getClusterName() {
@@ -271,6 +283,7 @@ public void configure(Builder<String, Object> settings) {
 		settings.put(IndexClientFactory.MAX_TERMS_COUNT_KEY, ""+getMaxTermsCount());
 		settings.put(IndexClientFactory.TRANSLOG_SYNC_INTERVAL_KEY, getCommitInterval());
 		settings.put(IndexClientFactory.COMMIT_CONCURRENCY_LEVEL, getCommitConcurrencyLevel());
+		settings.put(IndexClientFactory.INDEX_BY_QUERY_CONCURRENCY_LEVEL, getIndexByQueryConcurrencyLevel());
 		settings.put(IndexClientFactory.CONNECT_TIMEOUT, getConnectTimeout());
 		settings.put(IndexClientFactory.SOCKET_TIMEOUT, getSocketTimeout());
 		settings.put(IndexClientFactory.CLUSTER_HEALTH_TIMEOUT, getClusterHealthTimeout());