Added another test for upsert action without prior create action

Signed-off-by: Krishna Kondaka <[email protected]>
opensearch-project · Oct 5, 2023 · e112319 · e112319
1 parent b2e9d27
commit e112319
Showing 1 changed file with 35 additions and 0 deletions.
diff --git a/...grationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkIT.java b/...grationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkIT.java
@@ -824,6 +824,41 @@ public void testBulkActionUpsertWithActions() throws IOException, InterruptedExc
         sink.shutdown();
     }
 
+    @Test
+    public void testBulkActionUpsertWithoutCreate() throws IOException, InterruptedException {
+        final String testIndexAlias = "test-alias-upd2";
+        final String testTemplateFile = Objects.requireNonNull(
+                getClass().getClassLoader().getResource(TEST_TEMPLATE_BULK_FILE)).getFile();
+
+        final String testIdField = "someId";
+        final String testId = "foo";
+        List<Record<Event>> testRecords = Collections.singletonList(jsonStringToRecord(generateCustomRecordJson3(testIdField, testId, "name", "value1", "newKey", "newValue")));
+        final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFile);
+        pluginSetting.getSettings().put(IndexConfiguration.DOCUMENT_ID_FIELD, testIdField);
+        List<Map<String, Object>> aList = new ArrayList<>();
+        Map<String, Object> aMap = new HashMap<>();
+        aMap.put("type", BulkAction.UPSERT.toString());
+        aList.add(aMap);
+        pluginSetting.getSettings().put(IndexConfiguration.ACTIONS, aList);
+        OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+        sink.output(testRecords);
+        List<Map<String, Object>> retSources = getSearchResponseDocSources(testIndexAlias);
+
+        MatcherAssert.assertThat(retSources.size(), equalTo(1));
+        Map<String, Object> source = retSources.get(0);
+        MatcherAssert.assertThat((String)source.get("name"), equalTo("value1"));
+        MatcherAssert.assertThat((String)source.get("newKey"), equalTo("newValue"));
+        MatcherAssert.assertThat(getDocumentCount(testIndexAlias, "_id", testId), equalTo(Integer.valueOf(1)));
+        sink.shutdown();
+        // verify metrics
+        final List<Measurement> bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
+                new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+                        .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
+        MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
+        // COUNT
+        Assert.assertEquals(1.0, bulkRequestLatencies.get(0).getValue(), 0);
+    }
+
     @Test
     public void testBulkActionDeleteWithActions() throws IOException, InterruptedException {
         final String testIndexAlias = "test-alias-upd1";