Merge pull request #564 from rahul-tarento/course-batch-refactoring

es sync changes for bulk insert and async read
Sunbird-Lern · Jul 30, 2019 · caa7f84 · caa7f84
2 parents 7237ea1 + 536b158
commit caa7f84
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 8 deletions.
diff --git a/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandraimpl/CassandraDACImpl.java b/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandraimpl/CassandraDACImpl.java
@@ -9,6 +9,7 @@
 import com.google.common.util.concurrent.Futures;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executors;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.collections4.MapUtils;
 import org.sunbird.common.CassandraUtil;
@@ -84,7 +85,7 @@ public void applyOperationOnRecordsAsync(
         }
       }
       ResultSetFuture future = session.executeAsync(select);
-      Futures.addCallback(future, callback);
+      Futures.addCallback(future, callback, Executors.newFixedThreadPool(1));
     } catch (Exception e) {
       ProjectLogger.log(Constants.EXCEPTION_MSG_FETCH + table + " : " + e.getMessage(), e);
       throw new ProjectCommonException(

diff --git a/sunbird-es-utils/src/main/java/org/sunbird/common/ElasticSearchRestHighImpl.java b/sunbird-es-utils/src/main/java/org/sunbird/common/ElasticSearchRestHighImpl.java
@@ -547,7 +547,7 @@ public Future<Boolean> bulkInsert(String index, List<Map<String, Object>> dataLi
     BulkRequest request = new BulkRequest();
     Promise<Boolean> promise = Futures.promise();
     for (Map<String, Object> data : dataList) {
-      request.add(new IndexRequest(index, _DOC).source(data));
+      request.add(new IndexRequest(index, _DOC, (String) data.get(JsonKey.ID)).source(data));
     }
     ActionListener<BulkResponse> listener =
         new ActionListener<BulkResponse>() {
@@ -559,12 +559,15 @@ public void onResponse(BulkResponse bulkResponse) {
               while (responseItr.hasNext()) {
 
                 BulkItemResponse bResponse = responseItr.next();
-                ProjectLogger.log(
-                    "ElasticSearchRestHighImpl:bulkinsert: api response==="
-                        + bResponse.getId()
-                        + " "
-                        + bResponse.isFailed(),
-                    LoggerEnum.INFO.name());
+
+                if (bResponse.isFailed()) {
+                  ProjectLogger.log(
+                      "ElasticSearchRestHighImpl:bulkinsert: api response==="
+                          + bResponse.getId()
+                          + " "
+                          + bResponse.getFailureMessage(),
+                      LoggerEnum.INFO.name());
+                }
               }
             }
           }