Reapply "Run tests on the current JVM for Java 17 & 21 / Gradle 8.8 (#…

…4730)" (#4762) This reverts commit 5c7d58c. Signed-off-by: David Venable <[email protected]>
opensearch-project · Jul 29, 2024 · da5e480 · da5e480
1 parent 8f33225
commit da5e480
Show file tree

Hide file tree

Showing 12 changed files with 72 additions and 50 deletions.
diff --git a/build.gradle b/build.gradle
@@ -226,6 +226,9 @@ subprojects {
 
     test {
         useJUnitPlatform()
+        javaLauncher = javaToolchains.launcherFor {
+            languageVersion = JavaLanguageVersion.current()
+        }
         reports {
             junitXml.required
             html.required

diff --git a/data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/AbstractSink.java b/data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/AbstractSink.java
@@ -28,6 +28,7 @@ public abstract class AbstractSink<T extends Record<?>> implements Sink<T> {
     private Thread retryThread;
     private int maxRetries;
     private int waitTimeMs;
+    private SinkThread sinkThread;
 
     public AbstractSink(final PluginSetting pluginSetting, int numRetries, int waitTimeMs) {
         this.pluginMetrics = PluginMetrics.fromPluginSetting(pluginSetting);
@@ -51,7 +52,8 @@ public void initialize() {
         // the exceptions which are not retryable.
         doInitialize();
         if (!isReady() && retryThread == null) {
-            retryThread = new Thread(new SinkThread(this, maxRetries, waitTimeMs));
+            sinkThread = new SinkThread(this, maxRetries, waitTimeMs);
+            retryThread = new Thread(sinkThread);
             retryThread.start();
         }
     }
@@ -76,7 +78,7 @@ public void output(Collection<T> records) {
     @Override
     public void shutdown() {
         if (retryThread != null) {
-            retryThread.stop();
+            sinkThread.stop();
         }
     }
 

diff --git a/data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/SinkThread.java b/data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/SinkThread.java
@@ -10,6 +10,8 @@ class SinkThread implements Runnable {
     private int maxRetries;
     private int waitTimeMs;
 
+    private volatile boolean isStopped = false;
+
     public SinkThread(AbstractSink sink, int maxRetries, int waitTimeMs) {
         this.sink = sink;
         this.maxRetries = maxRetries;
@@ -19,11 +21,15 @@ public SinkThread(AbstractSink sink, int maxRetries, int waitTimeMs) {
     @Override
     public void run() {
         int numRetries = 0;
-        while (!sink.isReady() && numRetries++ < maxRetries) {
+        while (!sink.isReady() && numRetries++ < maxRetries && !isStopped) {
             try {
                 Thread.sleep(waitTimeMs);
                 sink.doInitialize();
             } catch (InterruptedException e){}
         }
     }
+
+    public void stop() {
+        isStopped = true;
+    }
 }
diff --git a/data-prepper-api/src/test/java/org/opensearch/dataprepper/model/sink/AbstractSinkTest.java b/data-prepper-api/src/test/java/org/opensearch/dataprepper/model/sink/AbstractSinkTest.java
@@ -11,15 +11,10 @@
 import org.opensearch.dataprepper.metrics.MetricNames;
 import org.opensearch.dataprepper.metrics.MetricsTestUtil;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
-import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.model.event.Event;
-import org.opensearch.dataprepper.model.event.JacksonEvent;
 import org.opensearch.dataprepper.model.event.EventHandle;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.when;
-import static org.mockito.Mockito.mock;
+import org.opensearch.dataprepper.model.event.JacksonEvent;
+import org.opensearch.dataprepper.model.record.Record;
 
 import java.time.Duration;
 import java.util.Arrays;
@@ -30,6 +25,12 @@
 import java.util.UUID;
 
 import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 class AbstractSinkTest {
     private int count;
@@ -71,13 +72,13 @@ void testMetrics() {
     }
 
     @Test
-    void testSinkNotReady() {
+    void testSinkNotReady() throws InterruptedException {
         final String sinkName = "testSink";
         final String pipelineName = "pipelineName";
         MetricsTestUtil.initMetrics();
         PluginSetting pluginSetting = new PluginSetting(sinkName, Collections.emptyMap());
         pluginSetting.setPipelineName(pipelineName);
-        AbstractSink<Record<String>> abstractSink = new AbstractSinkNotReadyImpl(pluginSetting);
+        AbstractSinkNotReadyImpl abstractSink = new AbstractSinkNotReadyImpl(pluginSetting);
         abstractSink.initialize();
         assertEquals(abstractSink.isReady(), false);
         assertEquals(abstractSink.getRetryThreadState(), Thread.State.RUNNABLE);
@@ -87,7 +88,10 @@ void testSinkNotReady() {
         await().atMost(Duration.ofSeconds(5))
                 .until(abstractSink::isReady);
         assertEquals(abstractSink.getRetryThreadState(), Thread.State.TERMINATED);
+        int initCountBeforeShutdown = abstractSink.initCount;
         abstractSink.shutdown();
+        Thread.sleep(200);
+        assertThat(abstractSink.initCount, equalTo(initCountBeforeShutdown));
     }
 
     @Test

diff --git a/data-prepper-core/build.gradle b/data-prepper-core/build.gradle
@@ -48,7 +48,6 @@ dependencies {
         exclude group: 'commons-logging', module: 'commons-logging'
     }
     implementation 'software.amazon.cloudwatchlogs:aws-embedded-metrics:2.0.0-beta-1'
-    testImplementation 'org.apache.logging.log4j:log4j-jpl:2.23.0'
     testImplementation testLibs.spring.test
     implementation libs.armeria.core
     implementation libs.armeria.grpc
@@ -89,8 +88,6 @@ task integrationTest(type: Test) {
 
     classpath = sourceSets.integrationTest.runtimeClasspath
 
-    systemProperty 'log4j.configurationFile', 'src/test/resources/log4j2.properties'
-
     filter {
         includeTestsMatching '*IT'
     }

diff --git a/...vro-codecs/src/test/java/org/opensearch/dataprepper/avro/AvroAutoSchemaGeneratorTest.java b/...vro-codecs/src/test/java/org/opensearch/dataprepper/avro/AvroAutoSchemaGeneratorTest.java
@@ -17,7 +17,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Random;
+import java.util.Timer;
 import java.util.UUID;
 import java.util.stream.Stream;
 
@@ -218,7 +218,7 @@ static class SomeUnknownTypesArgumentsProvider implements ArgumentsProvider {
         @Override
         public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
             return Stream.of(
-                    arguments(Random.class),
+                    arguments(Timer.class),
                     arguments(InputStream.class),
                     arguments(File.class)
             );

diff --git a/...st/java/org/opensearch/dataprepper/plugins/buffer/blockingbuffer/BlockingBufferTests.java b/...st/java/org/opensearch/dataprepper/plugins/buffer/blockingbuffer/BlockingBufferTests.java
@@ -328,7 +328,7 @@ public Stream<? extends Arguments> provideArguments(final ExtensionContext conte
             return Stream.of(
                     Arguments.of(0, randomInt + 1, 0.0),
                     Arguments.of(1, 100, 1.0),
-                    Arguments.of(randomInt, randomInt, 100.0),
+                    Arguments.of(randomInt + 1, randomInt + 1, 100.0),
                     Arguments.of(randomInt, randomInt + 250, ((double) randomInt / (randomInt + 250)) * 100),
                     Arguments.of(6, 9, 66.66666666666666),
                     Arguments.of(531, 1000, 53.1),

diff --git a/...est/java/org/opensearch/dataprepper/plugins/codec/event_json/EventJsonInputCodecTest.java b/...est/java/org/opensearch/dataprepper/plugins/codec/event_json/EventJsonInputCodecTest.java
@@ -11,9 +11,12 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
+
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.mock;
+
 import org.mockito.Mock;
+
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.CoreMatchers.not;
@@ -28,6 +31,7 @@
 import java.io.ByteArrayInputStream;
 
 import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.List;
 import java.util.LinkedList;
 import java.util.Map;
@@ -56,7 +60,7 @@ public EventJsonInputCodec createInputCodec() {
     @ParameterizedTest
     @ValueSource(strings = {"", "{}"})
     public void emptyTest(String input) throws Exception {
-        input = "{\""+EventJsonDefines.VERSION+"\":\""+DataPrepperVersion.getCurrentVersion().toString()+"\", \""+EventJsonDefines.EVENTS+"\":["+input+"]}";
+        input = "{\"" + EventJsonDefines.VERSION + "\":\"" + DataPrepperVersion.getCurrentVersion().toString() + "\", \"" + EventJsonDefines.EVENTS + "\":[" + input + "]}";
         ByteArrayInputStream inputStream = new ByteArrayInputStream(input.getBytes());
         inputCodec = createInputCodec();
         Consumer<Record<Event>> consumer = mock(Consumer.class);
@@ -70,15 +74,15 @@ public void inCompatibleVersionTest() throws Exception {
         final String key = UUID.randomUUID().toString();
         final String value = UUID.randomUUID().toString();
         Map<String, Object> data = Map.of(key, value);
-        Instant startTime = Instant.now();
+        Instant startTime = Instant.now().truncatedTo(ChronoUnit.MICROS);
         Event event = createEvent(data, startTime);
 
         Map<String, Object> dataMap = event.toMap();
         Map<String, Object> metadataMap = objectMapper.convertValue(event.getMetadata(), Map.class);
-        String input = "{\""+EventJsonDefines.VERSION+"\":\"3.0\", \""+EventJsonDefines.EVENTS+"\":[";
+        String input = "{\"" + EventJsonDefines.VERSION + "\":\"3.0\", \"" + EventJsonDefines.EVENTS + "\":[";
         String comma = "";
         for (int i = 0; i < 2; i++) {
-            input += comma+"{\"data\":"+objectMapper.writeValueAsString(dataMap)+","+"\"metadata\":"+objectMapper.writeValueAsString(metadataMap)+"}";
+            input += comma + "{\"data\":" + objectMapper.writeValueAsString(dataMap) + "," + "\"metadata\":" + objectMapper.writeValueAsString(metadataMap) + "}";
             comma = ",";
         }
         input += "]}";
@@ -95,24 +99,24 @@ public void basicTest() throws Exception {
         final String key = UUID.randomUUID().toString();
         final String value = UUID.randomUUID().toString();
         Map<String, Object> data = Map.of(key, value);
-        Instant startTime = Instant.now();
+        Instant startTime = Instant.now().truncatedTo(ChronoUnit.MICROS);
         Event event = createEvent(data, startTime);
 
         Map<String, Object> dataMap = event.toMap();
         Map<String, Object> metadataMap = objectMapper.convertValue(event.getMetadata(), Map.class);
-        String input = "{\""+EventJsonDefines.VERSION+"\":\""+DataPrepperVersion.getCurrentVersion().toString()+"\", \""+EventJsonDefines.EVENTS+"\":[";
+        String input = "{\"" + EventJsonDefines.VERSION + "\":\"" + DataPrepperVersion.getCurrentVersion().toString() + "\", \"" + EventJsonDefines.EVENTS + "\":[";
         String comma = "";
         for (int i = 0; i < 2; i++) {
-            input += comma+"{\"data\":"+objectMapper.writeValueAsString(dataMap)+","+"\"metadata\":"+objectMapper.writeValueAsString(metadataMap)+"}";
+            input += comma + "{\"data\":" + objectMapper.writeValueAsString(dataMap) + "," + "\"metadata\":" + objectMapper.writeValueAsString(metadataMap) + "}";
             comma = ",";
         }
         input += "]}";
         inputStream = new ByteArrayInputStream(input.getBytes());
         List<Record<Event>> records = new LinkedList<>();
         inputCodec.parse(inputStream, records::add);
         assertThat(records.size(), equalTo(2));
-        for(Record record : records) {
-            Event e = (Event)record.getData();
+        for (Record record : records) {
+            Event e = (Event) record.getData();
             assertThat(e.get(key, String.class), equalTo(value));
             assertThat(e.getMetadata().getTimeReceived(), equalTo(startTime));
             assertThat(e.getMetadata().getTags().size(), equalTo(0));
@@ -126,24 +130,24 @@ public void test_with_timeReceivedOverridden() throws Exception {
         final String key = UUID.randomUUID().toString();
         final String value = UUID.randomUUID().toString();
         Map<String, Object> data = Map.of(key, value);
-        Instant startTime = Instant.now().minusSeconds(5);
+        Instant startTime = Instant.now().truncatedTo(ChronoUnit.MICROS).minusSeconds(5);
         Event event = createEvent(data, startTime);
 
         Map<String, Object> dataMap = event.toMap();
         Map<String, Object> metadataMap = objectMapper.convertValue(event.getMetadata(), Map.class);
-        String input = "{\""+EventJsonDefines.VERSION+"\":\""+DataPrepperVersion.getCurrentVersion().toString()+"\", \""+EventJsonDefines.EVENTS+"\":[";
+        String input = "{\"" + EventJsonDefines.VERSION + "\":\"" + DataPrepperVersion.getCurrentVersion().toString() + "\", \"" + EventJsonDefines.EVENTS + "\":[";
         String comma = "";
         for (int i = 0; i < 2; i++) {
-            input += comma+"{\"data\":"+objectMapper.writeValueAsString(dataMap)+","+"\"metadata\":"+objectMapper.writeValueAsString(metadataMap)+"}";
+            input += comma + "{\"data\":" + objectMapper.writeValueAsString(dataMap) + "," + "\"metadata\":" + objectMapper.writeValueAsString(metadataMap) + "}";
             comma = ",";
         }
         input += "]}";
         inputStream = new ByteArrayInputStream(input.getBytes());
         List<Record<Event>> records = new LinkedList<>();
         inputCodec.parse(inputStream, records::add);
         assertThat(records.size(), equalTo(2));
-        for(Record record : records) {
-            Event e = (Event)record.getData();
+        for (Record record : records) {
+            Event e = (Event) record.getData();
             assertThat(e.get(key, String.class), equalTo(value));
             assertThat(e.getMetadata().getTimeReceived(), not(equalTo(startTime)));
             assertThat(e.getMetadata().getTags().size(), equalTo(0));
@@ -159,7 +163,7 @@ private Event createEvent(final Map<String, Object> json, final Instant timeRece
         if (timeReceived != null) {
             logBuilder.withTimeReceived(timeReceived);
         }
-        final JacksonEvent event = (JacksonEvent)logBuilder.build();
+        final JacksonEvent event = (JacksonEvent) logBuilder.build();
 
         return event;
     }

diff --git a/...va/org/opensearch/dataprepper/plugins/codec/event_json/EventJsonInputOutputCodecTest.java b/...va/org/opensearch/dataprepper/plugins/codec/event_json/EventJsonInputOutputCodecTest.java
@@ -6,9 +6,12 @@
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.mock;
+
 import org.mockito.Mock;
+
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
 
@@ -22,6 +25,7 @@
 import org.opensearch.dataprepper.model.log.JacksonLog;
 
 import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.List;
 import java.util.LinkedList;
 import java.util.Map;
@@ -64,7 +68,7 @@ public void basicTest() throws Exception {
         final String value = UUID.randomUUID().toString();
         Map<String, Object> data = Map.of(key, value);
 
-        Instant startTime = Instant.now();
+        Instant startTime = Instant.now().truncatedTo(ChronoUnit.MICROS);
         Event event = createEvent(data, startTime);
         outputCodec = createOutputCodec();
         inputCodec = createInputCodec();
@@ -75,8 +79,8 @@ public void basicTest() throws Exception {
         inputCodec.parse(new ByteArrayInputStream(outputStream.toByteArray()), records::add);
 
         assertThat(records.size(), equalTo(1));
-        for(Record record : records) {
-            Event e = (Event)record.getData();
+        for (Record record : records) {
+            Event e = (Event) record.getData();
             assertThat(e.get(key, String.class), equalTo(value));
             assertThat(e.getMetadata().getTimeReceived(), equalTo(startTime));
             assertThat(e.getMetadata().getTags().size(), equalTo(0));
@@ -90,7 +94,7 @@ public void multipleEventsTest() throws Exception {
         final String value = UUID.randomUUID().toString();
         Map<String, Object> data = Map.of(key, value);
 
-        Instant startTime = Instant.now();
+        Instant startTime = Instant.now().truncatedTo(ChronoUnit.MICROS);
         Event event = createEvent(data, startTime);
         outputCodec = createOutputCodec();
         inputCodec = createInputCodec();
@@ -103,8 +107,8 @@ public void multipleEventsTest() throws Exception {
         inputCodec.parse(new ByteArrayInputStream(outputStream.toByteArray()), records::add);
 
         assertThat(records.size(), equalTo(3));
-        for(Record record : records) {
-            Event e = (Event)record.getData();
+        for (Record record : records) {
+            Event e = (Event) record.getData();
             assertThat(e.get(key, String.class), equalTo(value));
             assertThat(e.getMetadata().getTimeReceived(), equalTo(startTime));
             assertThat(e.getMetadata().getTags().size(), equalTo(0));
@@ -122,7 +126,7 @@ public void extendedTest() throws Exception {
 
         Set<String> tags = Set.of(UUID.randomUUID().toString(), UUID.randomUUID().toString());
         List<String> tagsList = tags.stream().collect(Collectors.toList());
-        Instant startTime = Instant.now();
+        Instant startTime = Instant.now().truncatedTo(ChronoUnit.MICROS);
         Event event = createEvent(data, startTime);
         Instant origTime = startTime.minusSeconds(5);
         event.getMetadata().setExternalOriginationTime(origTime);
@@ -135,11 +139,11 @@ public void extendedTest() throws Exception {
         outputCodec.complete(outputStream);
         assertThat(outputCodec.getExtension(), equalTo(EventJsonOutputCodec.EVENT_JSON));
         List<Record<Event>> records = new LinkedList<>();
-inputCodec.parse(new ByteArrayInputStream(outputStream.toByteArray()), records::add);
+        inputCodec.parse(new ByteArrayInputStream(outputStream.toByteArray()), records::add);
 
         assertThat(records.size(), equalTo(1));
-        for(Record record : records) {
-            Event e = (Event)record.getData();
+        for (Record record : records) {
+            Event e = (Event) record.getData();
             assertThat(e.get(key, String.class), equalTo(value));
             assertThat(e.getMetadata().getTimeReceived(), equalTo(startTime));
             assertThat(e.getMetadata().getTags(), equalTo(tags));
@@ -157,7 +161,7 @@ private Event createEvent(final Map<String, Object> json, final Instant timeRece
         if (timeReceived != null) {
             logBuilder.withTimeReceived(timeReceived);
         }
-        final JacksonEvent event = (JacksonEvent)logBuilder.build();
+        final JacksonEvent event = (JacksonEvent) logBuilder.build();
 
         return event;
     }