Adding new bulk processor for elasticsearch implementation

Dockerfile

@@ -1,8 +1,8 @@
 FROM openjdk:11.0.8-jdk
 COPY . /mms
 WORKDIR /mms
-RUN ./gradlew --no-daemon build -x test
+RUN ./gradlew --no-daemon bootJar
 
 RUN cp /mms/example/build/libs/example*.jar /app.jar
-ENTRYPOINT ["java", "--add-opens", "java.base/java.lang=ALL-UNNAMED","-jar", "/app.jar"] 
+ENTRYPOINT ["java", "-Djdk.tls.client.protocols=TLSv1", "--add-opens", "java.base/java.lang=ALL-UNNAMED", "-jar", "/app.jar"]
 EXPOSE 8080

elastic/src/main/java/org/openmbee/sdvc/elastic/BaseElasticDAOImpl.java

@@ -7,8 +7,12 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.bulk.BackoffPolicy;
+import org.elasticsearch.action.bulk.BulkProcessor;
 import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.get.GetRequest;
 import org.elasticsearch.action.get.GetResponse;
@@ -21,16 +25,23 @@
 import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.get.GetResult;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 import org.openmbee.sdvc.elastic.utils.Index;
 import org.openmbee.sdvc.json.BaseJson;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
 
 public abstract class BaseElasticDAOImpl<E extends Map<String, Object>> {
 
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
     @Value("${elasticsearch.limit.result}")
     protected int resultLimit;
     @Value("${elasticsearch.limit.term}")
@@ -39,10 +50,10 @@ public abstract class BaseElasticDAOImpl<E extends Map<String, Object>> {
     protected RestHighLevelClient client;
     private static final RequestOptions REQUEST_OPTIONS;
     static {
-        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
+        RequestOptions.Builder requestBuilder = RequestOptions.DEFAULT.toBuilder();
         // TODO: Should be configureable
-        builder.setHttpAsyncResponseConsumerFactory(new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(1024 * 1024 * 1024));
-        REQUEST_OPTIONS = builder.build();
+        requestBuilder.setHttpAsyncResponseConsumerFactory(new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(1024 * 1024 * 1024));
+        REQUEST_OPTIONS = requestBuilder.build();
     }
 
     @Autowired
@@ -133,18 +144,21 @@ public List<E> findAllById(String index, Set<String> docIds) {
     }
 
     public void indexAll(String index, Collection<? extends BaseJson> jsons) {
+        BulkProcessor bulkProcessor = getBulkProcessor(client);
+        for (BaseJson json : jsons) {
+            bulkProcessor.add(new IndexRequest(index).id(json.getDocId()).source(json));
+        }
         try {
-            BulkRequest bulkIndex = new BulkRequest();
-            for (BaseJson json : jsons) {
-                bulkIndex.add(new IndexRequest(index).id(json.getDocId()).source(json));
+            if(!bulkProcessor.awaitClose(1200L, TimeUnit.SECONDS)) {
+                logger.error("Timed out in bulk processing");
             }
-            client.bulk(bulkIndex, REQUEST_OPTIONS);
-        } catch (IOException e) {
-            throw new RuntimeException(e);
+        } catch (InterruptedException e) {
+            logger.error("Index all interrupted: ", e);
         }
+
     }
 
-    public void index(String index, BaseJson json) {
+    public void index(String index, BaseJson<?> json) {
         try {
             client.index(new IndexRequest(index).id(json.getDocId()).source(json),
                 REQUEST_OPTIONS);
@@ -174,4 +188,37 @@ public E update(String index, BaseJson json) {
         }
         return response;
     }
+
+    private static BulkProcessor getBulkProcessor(RestHighLevelClient client) {
+        return getBulkProcessor(client, null);
+    }
+
+    private static BulkProcessor getBulkProcessor(RestHighLevelClient client,  BulkProcessor.Listener listener) {
+        if (listener == null) {
+            listener = new BulkProcessor.Listener() {
+                private final Logger logger = LoggerFactory.getLogger(getClass());
+                @Override
+                public void beforeBulk(long executionId, BulkRequest request) {
+                }
+
+                @Override
+                public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
+                }
+
+                @Override
+                public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
+                    logger.error("Error in bulk processing: ", failure);
+                }
+            };
+        }
+        BulkProcessor.Builder bpBuilder = BulkProcessor.builder((request, bulkListener) -> client
+            .bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener);
+        bpBuilder.setBulkActions(5000);
+        bpBuilder.setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB));
+        bpBuilder.setConcurrentRequests(1);
+        bpBuilder.setFlushInterval(TimeValue.timeValueSeconds(5));
+        bpBuilder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueMillis(100), 3));
+
+        return bpBuilder.build();
+    }
 }

example/example.gradle

@@ -18,13 +18,11 @@ dependencies {
     implementation(
         project(':authenticator'),
         project(':localuser'),
-        project(':ldap'),
         project(':cameo'),
         project(':elastic'),
         project(':jupyter'),
         project(':permissions'),
         project(':webhooks'),
-        project(':twc'),
         project(':search'),
         project(':artifacts'),
         'org.springframework.boot:spring-boot-starter-web',

example/src/main/java/org/openmbee/sdvc/example/config/ExampleSecurityConfig.java

@@ -1,7 +1,6 @@
 package org.openmbee.sdvc.example.config;
 
 import org.openmbee.sdvc.authenticator.config.AuthSecurityConfig;
-import org.openmbee.sdvc.twc.config.TwcAuthSecurityConfig;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -36,15 +35,11 @@ public class ExampleSecurityConfig extends WebSecurityConfigurerAdapter implemen
     @Autowired
     AuthSecurityConfig authSecurityConfig;
 
-    @Autowired
-    TwcAuthSecurityConfig twcAuthSecurityConfig;
-
     @Override
     public void configure(HttpSecurity http) throws Exception {
         http.csrf().disable().authorizeRequests().anyRequest().permitAll().and().httpBasic();
         http.headers().cacheControl();
         http.addFilterAfter(corsFilter(), ExceptionTranslationFilter.class);
-        twcAuthSecurityConfig.setAuthConfig(http);
         authSecurityConfig.setAuthConfig(http);
     }
 

gradle.properties

@@ -7,4 +7,4 @@ springSecurityVersion=5.3.1.RELEASE
 springDataVersion=2.2.6.RELEASE
 jacksonVersion=2.10.3
 log4jVersion=2.13.1
-elasticVersion=7.1.1
+elasticVersion=7.7.1

twc/src/main/java/org/openmbee/sdvc/twc/services/TwcRevisionMmsCommitMapService.java

@@ -36,7 +36,7 @@ public void setBranchRepository(BranchDAO branchRepository) {
      */
     public CommitsResponse updateTwcRevisionID(String projectId, String commitId, String revisionId) {
         CommitsResponse commitsResponse = new CommitsResponse();
-        if (revisionId.isEmpty() || revisionId.isBlank()) {
+        if (revisionId == null || revisionId.isEmpty()) {
             return commitsResponse.addMessage("Revision id can not be empty");
         }
         ContextHolder.setContext(projectId);