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

…4730)" (#4762) This reverts commit 67f3595. Signed-off-by: Hai Yan <[email protected]>
opensearch-project · Jul 23, 2024 · 5c7d58c · 5c7d58c
1 parent 427bb78
commit 5c7d58c
Show file tree

Hide file tree

Showing 12 changed files with 50 additions and 72 deletions.
diff --git a/build.gradle b/build.gradle
@@ -226,9 +226,6 @@ 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,7 +28,6 @@ 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);
@@ -52,8 +51,7 @@ public void initialize() {
         // the exceptions which are not retryable.
         doInitialize();
         if (!isReady() && retryThread == null) {
-            sinkThread = new SinkThread(this, maxRetries, waitTimeMs);
-            retryThread = new Thread(sinkThread);
+            retryThread = new Thread(new SinkThread(this, maxRetries, waitTimeMs));
             retryThread.start();
         }
     }
@@ -78,7 +76,7 @@ public void output(Collection<T> records) {
     @Override
     public void shutdown() {
         if (retryThread != null) {
-            sinkThread.stop();
+            retryThread.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,8 +10,6 @@ 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;
@@ -21,15 +19,11 @@ public SinkThread(AbstractSink sink, int maxRetries, int waitTimeMs) {
     @Override
     public void run() {
         int numRetries = 0;
-        while (!sink.isReady() && numRetries++ < maxRetries && !isStopped) {
+        while (!sink.isReady() && numRetries++ < maxRetries) {
             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,10 +11,15 @@
 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.EventHandle;
 import org.opensearch.dataprepper.model.event.JacksonEvent;
-import org.opensearch.dataprepper.model.record.Record;
+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 java.time.Duration;
 import java.util.Arrays;
@@ -25,12 +30,6 @@
 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;
@@ -72,13 +71,13 @@ void testMetrics() {
     }
 
     @Test
-    void testSinkNotReady() throws InterruptedException {
+    void testSinkNotReady() {
         final String sinkName = "testSink";
         final String pipelineName = "pipelineName";
         MetricsTestUtil.initMetrics();
         PluginSetting pluginSetting = new PluginSetting(sinkName, Collections.emptyMap());
         pluginSetting.setPipelineName(pipelineName);
-        AbstractSinkNotReadyImpl abstractSink = new AbstractSinkNotReadyImpl(pluginSetting);
+        AbstractSink<Record<String>> abstractSink = new AbstractSinkNotReadyImpl(pluginSetting);
         abstractSink.initialize();
         assertEquals(abstractSink.isReady(), false);
         assertEquals(abstractSink.getRetryThreadState(), Thread.State.RUNNABLE);
@@ -88,10 +87,7 @@ void testSinkNotReady() throws InterruptedException {
         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,6 +48,7 @@ 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
@@ -88,6 +89,8 @@ 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.Timer;
+import java.util.Random;
 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(Timer.class),
+                    arguments(Random.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 + 1, randomInt + 1, 100.0),
+                    Arguments.of(randomInt, randomInt, 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,12 +11,9 @@
 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;
@@ -31,7 +28,6 @@
 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;
@@ -60,7 +56,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);
@@ -74,15 +70,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().truncatedTo(ChronoUnit.MICROS);
+        Instant startTime = Instant.now();
         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 += "]}";
@@ -99,24 +95,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().truncatedTo(ChronoUnit.MICROS);
+        Instant startTime = Instant.now();
         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));
@@ -130,24 +126,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().truncatedTo(ChronoUnit.MICROS).minusSeconds(5);
+        Instant startTime = Instant.now().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));
@@ -163,7 +159,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,12 +6,9 @@
 
 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;
 
@@ -25,7 +22,6 @@
 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;
@@ -68,7 +64,7 @@ public void basicTest() throws Exception {
         final String value = UUID.randomUUID().toString();
         Map<String, Object> data = Map.of(key, value);
 
-        Instant startTime = Instant.now().truncatedTo(ChronoUnit.MICROS);
+        Instant startTime = Instant.now();
         Event event = createEvent(data, startTime);
         outputCodec = createOutputCodec();
         inputCodec = createInputCodec();
@@ -79,8 +75,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));
@@ -94,7 +90,7 @@ public void multipleEventsTest() throws Exception {
         final String value = UUID.randomUUID().toString();
         Map<String, Object> data = Map.of(key, value);
 
-        Instant startTime = Instant.now().truncatedTo(ChronoUnit.MICROS);
+        Instant startTime = Instant.now();
         Event event = createEvent(data, startTime);
         outputCodec = createOutputCodec();
         inputCodec = createInputCodec();
@@ -107,8 +103,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));
@@ -126,7 +122,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().truncatedTo(ChronoUnit.MICROS);
+        Instant startTime = Instant.now();
         Event event = createEvent(data, startTime);
         Instant origTime = startTime.minusSeconds(5);
         event.getMetadata().setExternalOriginationTime(origTime);
@@ -139,11 +135,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));
@@ -161,7 +157,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;
     }