Adds support for bulk create option #1561

jzonthemtn · 2022-07-02T14:13:49Z

Description

Adds support for bulk create option

Issues Resolved

#1457 Create-only actions in OpenSearch bulk requests

Check List

New functionality includes testing.
New functionality has been documented.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

codecov-commenter · 2022-07-02T14:23:35Z

Codecov Report

Merging #1561 (6556c5b) into main (385cf6d) will increase coverage by 0.04%.
The diff coverage is n/a.

@@             Coverage Diff              @@
##               main    #1561      +/-   ##
============================================
+ Coverage     94.51%   94.55%   +0.04%     
- Complexity     1152     1236      +84     
============================================
  Files           158      162       +4     
  Lines          3297     3509     +212     
  Branches        268      290      +22     
============================================
+ Hits           3116     3318     +202     
- Misses          128      129       +1     
- Partials         53       62       +9

Impacted Files	Coverage Δ
...dataprepper/model/configuration/PipelineModel.java	`100.00% <0.00%> (ø)`
...h/dataprepper/logstash/mapping/LogstashMapper.java	`100.00% <0.00%> (ø)`
...pper/peerforwarder/PeerForwarderConfiguration.java	`84.52% <0.00%> (ø)`
...eerforwarder/PeerForwardingProcessorDecorator.java	`100.00% <0.00%> (ø)`
...prepper/peerforwarder/discovery/DiscoveryMode.java	`100.00% <0.00%> (ø)`
...aprepper/peerforwarder/PeerForwarderAppConfig.java	`100.00% <0.00%> (ø)`
.../com/amazon/dataprepper/parser/PipelineParser.java	`93.46% <0.00%> (+3.77%)`	⬆️
...prepper/parser/model/DataPrepperConfiguration.java	`93.75% <0.00%> (+8.96%)`	⬆️

Help us with your feedback. Take ten seconds to tell us how you rate us.

graytaylor0

This looks great! Thanks for making this contribution. I had a couple comments for some general code improvements, but nothing major.

graytaylor0 · 2022-07-06T22:02:09Z

.../opensearch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

-    if (bulkRequest.getOperationsCount() > 0) {
-      flushBatch(bulkRequest);
+    } else {
+      throw new RuntimeException("Invalid action: " + action);


We should never have to throw a RuntimeException for an invalid configuration. I think it would be best to create an enum for the possible BulkActions, and to have an IllegalArgumentException be thrown from the IndexConfiguration class when trying to convert the action String passed by the user to the enum type

graytaylor0 · 2022-07-07T00:17:08Z

...h/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/index/IndexConfiguration.java

@@ -260,6 +273,11 @@ public Builder withIsmPolicyFile(final String ismPolicyFile) {
            return this;
        }

+        public Builder withAction(final String action) {
+            this.action = action;


We can add validation on construction by adding the following line here (given that BulkActions is an enum)

checkArgument(EnumUtils.isValidEnum(BulkActions.class, action), "action must be one of the folllowing: BulkActions.values()")

graytaylor0 · 2022-07-07T15:09:19Z

.../opensearch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

+        flushBatch(bulkRequest);
+      }
+
+    } else if(IndexConfiguration.BULK_ACTION_INDEX.equals(action)) {


We can limit the amount of duplicate code by refactoring the if create, else if index to look something like this. It would also be nice to move creation of the BulkOperation to a Factory class with functions of provideBulkOperationForCreate and provideBulkOperationForIndex, but that isn't necessary for this PR.

for (final Record<Object> record : records) { final SerializedJson document = getDocument(record.getData()); final Optional<String> docId = getDocumentIdFromDocument(document); BulkOperation bulkOperation; if (action.equals("index")) { final IndexOperation.Builder<Object> indexOperationBuilder = new IndexOperation.Builder<>() .index(indexManager.getIndexAlias()); .document(document); if (docId.isPresent()) { indexOperationBuilder.id(docId); } indexOperationBuilder.build(); bulkOperation = new BulkOperation.Builder() .index(indexOperationBuilder.build()) .build(); } else if (action.equals("create")) { final CreateOperation.Builder<Object> createOperationBuilder = new CreateOperation.Builder<>() .index(indexManager.getIndexAlias()); .document(document); if (docId.isPresent()) { createOperationBuilder.id(docId); } createOperationBuilder.build(); bulkOperation = new BulkOperation.Builder() .create(createOperationsBuilder.build()) .build(); } final long estimatedBytesBeforeAdd = bulkRequest.estimateSizeInBytesWithDocument(bulkOperation); if (bulkSize >= 0 && estimatedBytesBeforeAdd >= bulkSize && bulkRequest.getOperationsCount() > 0) { flushBatch(bulkRequest); bulkRequest = bulkRequestSupplier.get(); } bulkRequest.addOperation(bulkOperation); } // Flush the remaining requests if (bulkRequest.getOperationsCount() > 0) { flushBatch(bulkRequest); } } } private Optional<String> getDocumentIdFromDocument(final SerializedJson document) { final Map documentAsMap; try { documentAsMap = objectMapper.readValue(document.getSerializedJson(), Map.class); } catch (IOException e) { throw new RuntimeException(e); } if (documentAsMap != null) { final String docId = (String) documentAsMap.get(documentIdField); if (docId != null) { return Optional.of(docId); } } return Optional.empty(); }

jzonthemtn · 2022-07-11T13:17:45Z

This looks great! Thanks for making this contribution. I had a couple comments for some general code improvements, but nothing major.

Awesome - thanks! I will get those addressed shortly.

jzonthemtn · 2022-07-11T14:43:07Z

@graytaylor0 Thanks a lot for those comments. Everything made sense. I pushed changes up.

dlvenable

Thanks @jzonthemtn for this contribution! I have one request regarding the naming in the pipeline configurations.

...pensearch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/bulk/BulkActions.java

jzonthemtn · 2022-07-11T21:24:18Z

Looks like I messed up in there by merging from main and not signing the merge commit. What's the best way to fix that?

graytaylor0 · 2022-07-11T21:53:19Z

Looks like I messed up in there by merging from main and not signing the merge commit. What's the best way to fix that?

Thanks for making those changes! The best way to fix the DCO will be to rebase your branch against main with

git rebase -i main, and then you can just squash all your commits (or at least the one without the DCO signing)

Signed-off-by: jzonthemtn <[email protected]>

dlvenable

Thanks @jzonthemtn for this contribution! And thank you for the quick responses to our feedback.

graytaylor0 · 2022-07-30T02:00:40Z

@jzonthemtn Any reason this was closed? If it was an accident we can reopen and merge.

jzonthemtn · 2022-07-30T11:36:11Z

@jzonthemtn Any reason this was closed? If it was an accident we can reopen and merge.

No, I don't know why. I can only guess I was in the wrong issue. Sorry for my clumsiness.

…ct#1561) Signed-off-by: jzonthemtn <[email protected]>

jzonthemtn requested a review from a team as a code owner July 2, 2022 14:13

graytaylor0 reviewed Jul 7, 2022

View reviewed changes

dlvenable requested changes Jul 11, 2022

View reviewed changes

...pensearch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/bulk/BulkActions.java Outdated Show resolved Hide resolved

#1457: Adding bulk create option.

6556c5b

Signed-off-by: jzonthemtn <[email protected]>

dlvenable approved these changes Jul 12, 2022

View reviewed changes

graytaylor0 approved these changes Jul 12, 2022

View reviewed changes

jzonthemtn closed this Jul 21, 2022

jzonthemtn deleted the 1457_create-only-actions branch July 21, 2022 14:47

jzonthemtn restored the 1457_create-only-actions branch July 30, 2022 11:35

jzonthemtn reopened this Jul 30, 2022

graytaylor0 merged commit 0ea2657 into opensearch-project:main Jul 30, 2022

dlvenable mentioned this pull request Aug 8, 2022

Create-only actions in OpenSearch bulk requests #1457

Closed

engechas pushed a commit to engechas/data-prepper that referenced this pull request Sep 12, 2022

opensearch-project#1457: Adding bulk create option. (opensearch-proje…

0c660cb

…ct#1561) Signed-off-by: jzonthemtn <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adds support for bulk create option #1561

Adds support for bulk create option #1561

jzonthemtn commented Jul 2, 2022 •

edited

Loading

codecov-commenter commented Jul 2, 2022 •

edited

Loading

graytaylor0 left a comment

graytaylor0 Jul 6, 2022

graytaylor0 Jul 7, 2022

graytaylor0 Jul 7, 2022

jzonthemtn commented Jul 11, 2022

jzonthemtn commented Jul 11, 2022

dlvenable left a comment

jzonthemtn commented Jul 11, 2022

graytaylor0 commented Jul 11, 2022

dlvenable left a comment

graytaylor0 commented Jul 30, 2022

jzonthemtn commented Jul 30, 2022

Adds support for bulk create option #1561

Adds support for bulk create option #1561

Conversation

jzonthemtn commented Jul 2, 2022 • edited Loading

Description

Issues Resolved

Check List

codecov-commenter commented Jul 2, 2022 • edited Loading

Codecov Report

graytaylor0 left a comment

Choose a reason for hiding this comment

graytaylor0 Jul 6, 2022

Choose a reason for hiding this comment

graytaylor0 Jul 7, 2022

Choose a reason for hiding this comment

graytaylor0 Jul 7, 2022

Choose a reason for hiding this comment

jzonthemtn commented Jul 11, 2022

jzonthemtn commented Jul 11, 2022

dlvenable left a comment

Choose a reason for hiding this comment

jzonthemtn commented Jul 11, 2022

graytaylor0 commented Jul 11, 2022

dlvenable left a comment

Choose a reason for hiding this comment

graytaylor0 commented Jul 30, 2022

jzonthemtn commented Jul 30, 2022

jzonthemtn commented Jul 2, 2022 •

edited

Loading

codecov-commenter commented Jul 2, 2022 •

edited

Loading