Skip to content

Commit f00a885

Browse files
Davidding4718Davidding4718
committed
Add integration test verifying processors work with empty, null, and bare key configurations
1 parent 27ffbe9 commit f00a885

4 files changed

Lines changed: 212 additions & 0 deletions

File tree

‎data-prepper-core/src/integrationTest/java/org/opensearch/dataprepper/integration/EmptyProcessorIT.java‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
package org.opensearch.dataprepper.integration;
11+
12+
import org.junit.jupiter.api.AfterEach;
13+
import org.junit.jupiter.api.BeforeEach;
14+
import org.junit.jupiter.api.Test;
15+
import org.opensearch.dataprepper.model.event.Event;
16+
import org.opensearch.dataprepper.model.event.JacksonEvent;
17+
import org.opensearch.dataprepper.model.record.Record;
18+
import org.opensearch.dataprepper.plugins.EmptyProcessor;
19+
import org.opensearch.dataprepper.plugins.InMemorySinkAccessor;
20+
import org.opensearch.dataprepper.plugins.InMemorySourceAccessor;
21+
import org.opensearch.dataprepper.test.framework.DataPrepperTestRunner;
22+
23+
import java.util.Collections;
24+
import java.util.List;
25+
import java.util.UUID;
26+
import java.util.concurrent.TimeUnit;
27+
import java.util.stream.Collectors;
28+
import java.util.stream.IntStream;
29+
30+
import static org.awaitility.Awaitility.await;
31+
import static org.hamcrest.CoreMatchers.equalTo;
32+
import static org.hamcrest.CoreMatchers.not;
33+
import static org.hamcrest.MatcherAssert.assertThat;
34+
import static org.hamcrest.Matchers.empty;
35+
36+
/**
37+
* Integration test verifying that a processor with no mandatory configuration
38+
* works correctly when configured as:
39+
* <ul>
40+
* <li>bare key (empty): {@code empty_processor:}</li>
41+
* <li>explicit null: {@code empty_processor: null}</li>
42+
* <li>explicit empty string: {@code empty_processor: ''}</li>
43+
* </ul>
44+
*
45+
* The pipeline configures 3 instances of {@code empty_processor}. Each instance
46+
* increments a counter on every event it processes. After all processors run,
47+
* the counter should equal 3 — proving all three configuration forms produce
48+
* valid, functioning processor instances.
49+
*
50+
* @see EmptyProcessor
51+
*/
52+
class EmptyProcessorIT {
53+
private static final String IN_MEMORY_IDENTIFIER = "EmptyProcessorIT";
54+
private static final String PIPELINE_CONFIGURATION_UNDER_TEST = "empty-processor-pipeline.yaml";
55+
private DataPrepperTestRunner dataPrepperTestRunner;
56+
private InMemorySourceAccessor inMemorySourceAccessor;
57+
private InMemorySinkAccessor inMemorySinkAccessor;
58+
59+
@BeforeEach
60+
void setUp() {
61+
dataPrepperTestRunner = DataPrepperTestRunner.builder()
62+
.withPipelinesDirectoryOrFile(PIPELINE_CONFIGURATION_UNDER_TEST)
63+
.build();
64+
65+
dataPrepperTestRunner.start();
66+
inMemorySourceAccessor = dataPrepperTestRunner.getInMemorySourceAccessor();
67+
inMemorySinkAccessor = dataPrepperTestRunner.getInMemorySinkAccessor();
68+
}
69+
70+
@AfterEach
71+
void tearDown() {
72+
dataPrepperTestRunner.stop();
73+
}
74+
75+
@Test
76+
void pipeline_with_single_record_has_count_equal_to_3() {
77+
final Event event = JacksonEvent.fromMessage(UUID.randomUUID().toString());
78+
final Record<Event> eventRecord = new Record<>(event);
79+
80+
inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, Collections.singletonList(eventRecord));
81+
82+
await().atMost(2, TimeUnit.SECONDS)
83+
.untilAsserted(() -> {
84+
assertThat(inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER), not(empty()));
85+
});
86+
87+
final List<Record<Event>> records = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
88+
assertThat(records.size(), equalTo(1));
89+
90+
final Integer count = records.get(0).getData().get(EmptyProcessor.COUNT_KEY, Integer.class);
91+
assertThat("All 3 empty_processor instances (bare key, null, empty string) should have processed the event",
92+
count, equalTo(3));
93+
}
94+
95+
@Test
96+
void pipeline_with_single_batch_of_records() {
97+
final int recordsToCreate = 200;
98+
final List<Record<Event>> inputRecords = IntStream.range(0, recordsToCreate)
99+
.mapToObj(i -> UUID.randomUUID().toString())
100+
.map(JacksonEvent::fromMessage)
101+
.map(Record::new)
102+
.collect(Collectors.toList());
103+
104+
inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, inputRecords);
105+
106+
await().atMost(2, TimeUnit.SECONDS)
107+
.untilAsserted(() -> {
108+
assertThat(inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER), not(empty()));
109+
});
110+
111+
final List<Record<Event>> sinkRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
112+
assertThat(sinkRecords.size(), equalTo(recordsToCreate));
113+
114+
for (final Record<Event> record : sinkRecords) {
115+
final Integer count = record.getData().get(EmptyProcessor.COUNT_KEY, Integer.class);
116+
assertThat("Each record should be processed by all 3 empty_processor instances",
117+
count, equalTo(3));
118+
}
119+
}
120+
}

‎data-prepper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/EmptyProcessor.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
package org.opensearch.dataprepper.plugins;
11+
12+
import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
13+
import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
14+
import org.opensearch.dataprepper.model.event.Event;
15+
import org.opensearch.dataprepper.model.processor.Processor;
16+
import org.opensearch.dataprepper.model.record.Record;
17+
18+
import java.util.Collection;
19+
20+
/**
21+
* A test processor with no mandatory configuration parameters.
22+
* Each instance increments a "empty_processor_count" field on the event.
23+
* <p>
24+
* This processor is used in integration tests to verify that plugins
25+
* configured as a bare key, explicit null, or explicit empty string
26+
* all result in valid processor instances that process events.
27+
* <p>
28+
* If 3 instances are configured in the pipeline, after processing
29+
* the "empty_processor_count" field on each event should equal 3.
30+
*/
31+
@DataPrepperPlugin(name = "empty_processor", pluginType = Processor.class, pluginConfigurationType = EmptyProcessorConfig.class)
32+
public class EmptyProcessor implements Processor<Record<Event>, Record<Event>> {
33+
public static final String COUNT_KEY = "empty_processor_count";
34+
35+
@DataPrepperPluginConstructor
36+
public EmptyProcessor(final EmptyProcessorConfig config) {
37+
}
38+
39+
@Override
40+
public Collection<Record<Event>> execute(final Collection<Record<Event>> records) {
41+
for (final Record<Event> record : records) {
42+
final Integer currentCount = record.getData().get(COUNT_KEY, Integer.class);
43+
record.getData().put(COUNT_KEY, (currentCount == null ? 0 : currentCount) + 1);
44+
}
45+
return records;
46+
}
47+
48+
@Override
49+
public void prepareForShutdown() {
50+
}
51+
52+
@Override
53+
public boolean isReadyForShutdown() {
54+
return true;
55+
}
56+
57+
@Override
58+
public void shutdown() {
59+
}
60+
}

‎data-prepper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/EmptyProcessorConfig.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
package org.opensearch.dataprepper.plugins;
11+
12+
/**
13+
* Configuration for {@link EmptyProcessor}.
14+
* This processor has no mandatory configuration parameters,
15+
* allowing it to be configured as a bare key, null, or empty string.
16+
*/
17+
public class EmptyProcessorConfig {
18+
}

‎data-prepper-core/src/integrationTest/resources/org/opensearch/dataprepper/pipeline/empty-processor-pipeline.yaml‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
empty-processor-pipeline:
2+
delay: 10
3+
source:
4+
in_memory:
5+
testing_key: EmptyProcessorIT
6+
7+
processor:
8+
- empty_processor:
9+
- empty_processor: null
10+
- empty_processor: ''
11+
12+
sink:
13+
- in_memory:
14+
testing_key: EmptyProcessorIT

0 commit comments

Comments
 (0)