Implementation of conditional routing of sinks

[dlvenable]

Description

This PR implements conditional routing.

I've rebased this PR to include #1840 and now it works with multiple processor threads. Note that this only works for single-threaded pipelines due to #1189.

Issues Resolved

Resolves #1337

Testing Information

Sample Pipeline:

simple-test-pipeline:
  workers: 1
  delay: "5000"
  source:
    file:
      path: /usr/share/data-prepper-file-source/test-in.txt
      record_type: event
      format: json
  route:
    - alpha: '/value == "a"'
    - beta: '/value == "b"'
    - sigma: '/value == "s"'
  sink:
    - file:
        path: /usr/share/data-prepper-file-source/out-alpha.txt
        routes:
          - alpha
    - file:
        path: /usr/share/data-prepper-file-source/out-beta.txt
        routes:
          - beta
    - file:
        path: /usr/share/data-prepper-file-source/out-should-be-empty.txt
        routes:
          - sigma
    - file:
        path: /usr/share/data-prepper-file-source/out-alpha-beta.txt
        routes:
          - alpha
          - beta
    - file:
        path: /usr/share/data-prepper-file-source/out-all.txt

File Input Source:

cat file-source/test-in.txt
{"value":"a", "other":"x", "sequence" : 1}
{"value":"a", "other":"x", "sequence" : 2}
{"value":"b", "sequence" : 3}
{"value":"b", "sequence" : 4}
{"value":"c", "sequence" : 5}
{"value":"d", "sequence" : 6}
{"value":"b", "sequence" : 7}
{"value":"a", "sequence" : 8}

Run with Docker:

docker run --name data-prepper -p 4900:4900 \
  -v ${PWD}/pipelines.yaml:/usr/share/data-prepper/pipelines/pipelines.yaml \
  -v ${PWD}/file-source:/usr/share/data-prepper-file-source/ \
  opensearch-data-prepper:2.0.0-SNAPSHOT

Results:

tail -n +1 file-source/out-*
==> file-source/out-all.txt <==
{"value":"a","other":"x","sequence":1}
{"value":"a","other":"x","sequence":2}
{"value":"b","sequence":3}
{"value":"b","sequence":4}
{"value":"c","sequence":5}
{"value":"d","sequence":6}
{"value":"b","sequence":7}
{"value":"a","sequence":8}

==> file-source/out-alpha-beta.txt <==
{"value":"a","other":"x","sequence":1}
{"value":"a","other":"x","sequence":2}
{"value":"b","sequence":3}
{"value":"b","sequence":4}
{"value":"b","sequence":7}
{"value":"a","sequence":8}

==> file-source/out-alpha.txt <==
{"value":"a","other":"x","sequence":1}
{"value":"a","other":"x","sequence":2}
{"value":"a","sequence":8}

==> file-source/out-beta.txt <==
{"value":"b","sequence":3}
{"value":"b","sequence":4}
{"value":"b","sequence":7}

==> file-source/out-should-be-empty.txt <==

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.

[oeyh]

Works great! I was also able to run a small test with routes set on a pipeline connector.
A few small things below:

[oeyh]

Does this TODO item refer to the method defined around Line 249 and thus can be resolved?

[dlvenable]

Yes, that is correct. Thanks for noting this, and I will remove.

[oeyh]

nit:

Suggested change

	void getComponent_returns_input_routes() {
	void getRoutes_returns_input_routes() {

[oeyh]

Same here:

Suggested change

	void getComponent_returns_input_routes_when_empty() {
	void getRoutes_returns_input_routes_when_empty() {

[oeyh]

Pipeline filename should be valid_multiple_sinks_with_routes.yml.

[cmanning09]

This can be simplified by breaking this down. This would also eliminate variable re-assignment and reads better:

    if(data instanceof Event) {
        final Event event = (Event) data;
        final Set<String> matchedRoutes = findMatchedRoutes(event);
        recordsToRoutes.put(record, matchedRoutes);
    } else {
         nonEventRecords++;
        recordsToRoutes.put(record, Collections.emptySet());
    }

...

private Set<String> findMatchedRoutes(final Event event) {
    final Set<String> matchRoutes = new HashSet();
    for  (ConditionalRoute route : routes) {
       try {
           if (evaluator.evaluate(route.getCondition(), event)) {
               matchRoutes.add(route.getName());
           }
     } catch (final Exception ex) {
          LOG.error("Failed to evaluate route. This route will not be applied to any events.", ex);
     }
     return matchRoutes;
}

[cmanning09]

This can be simplified to eliminate the break. This should still short circuit and not loop through every item if a match occurs.

final Set<String> routesForEvent = recordsToRoutes
    .getOrDefault(event, Collections.emptySet());

final Set<String> dataFlowComponentRoutes =  dataFlowComponent.getRoutes();

if (routesForEvent.stream().anyMatch(dataFlowComponentRoutes::contains)) {
   recordsForComponent.add(event);
}