Inserting bulk clickthrough rates using an ActionListener.

o19s · Dec 10, 2024 · 5e388bb · 5e388bb
1 parent 0ca0822
commit 5e388bb
Showing 1 changed file with 20 additions and 7 deletions.
diff --git a/...ation-plugin/src/main/java/org/opensearch/eval/judgments/opensearch/OpenSearchHelper.java b/...ation-plugin/src/main/java/org/opensearch/eval/judgments/opensearch/OpenSearchHelper.java
@@ -30,14 +30,11 @@
 import java.io.IOException;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
-import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
-import java.util.TimeZone;
 import java.util.UUID;
 
 import static org.opensearch.eval.SearchQualityEvaluationPlugin.JUDGMENTS_INDEX_NAME;
@@ -263,8 +260,6 @@ public void indexClickthroughRates(final Map<String, Set<ClickthroughRate>> clic
 
         if(!clickthroughRates.isEmpty()) {
 
-            // TODO: Split this into multiple bulk insert requests.
-
             final BulkRequest request = new BulkRequest();
 
             for(final String userQuery : clickthroughRates.keySet()) {
@@ -278,15 +273,33 @@ public void indexClickthroughRates(final Map<String, Set<ClickthroughRate>> clic
                     jsonMap.put("ctr", clickthroughRate.getClickthroughRate());
                     jsonMap.put("object_id", clickthroughRate.getObjectId());
 
-                    final IndexRequest indexRequest = new IndexRequest(INDEX_QUERY_DOC_CTR).id(UUID.randomUUID().toString()).source(jsonMap);
+                    final IndexRequest indexRequest = new IndexRequest(INDEX_QUERY_DOC_CTR)
+                            .id(UUID.randomUUID().toString())
+                            .source(jsonMap);
 
                     request.add(indexRequest);
 
                 }
 
             }
 
-            client.bulk(request).get();
+            client.bulk(request, new ActionListener<>() {
+
+                @Override
+                public void onResponse(BulkResponse bulkItemResponses) {
+                    if(bulkItemResponses.hasFailures()) {
+                        LOGGER.error("Clickthrough rates were not all successfully indexed: {}", bulkItemResponses.buildFailureMessage());
+                    } else {
+                        LOGGER.debug("Clickthrough rates has been successfully indexed.");
+                    }
+                }
+
+                @Override
+                public void onFailure(Exception ex) {
+                    LOGGER.error("Indexing the clickthrough rates failed.", ex);
+                }
+
+            });
 
         }