Upgrade JDK toolchain/target version to 17 and add GraalJS

.github/workflows/CI.yml

@@ -6,7 +6,7 @@ on:
 
 env:
   python-version: '3.11'
-  java-version: '11'
+  java-version: '17'
   gradle-version: '8.0.2'
   node-version: '18.x'
 

.github/workflows/release-drafter.yml

@@ -6,7 +6,7 @@ on:
     tags:
       - "*"
 env:
-  java-version: '11'
+  java-version: '17'
   gradle-version: '8.0.2'
 
 permissions:

.github/workflows/sonar-qube.yml

@@ -5,7 +5,7 @@ on:
   pull_request:
 
 env:
-  java-version: '11'
+  java-version: '17'
   gradle-version: '8.0.2'
 
 jobs:
@@ -32,7 +32,7 @@ jobs:
       with:
         distribution: 'corretto'
         java-version: |
-            11
+            17
     - name: Cache SonarQube Scanner
       uses: actions/cache@v3
       with:

CreateSnapshot/build.gradle

@@ -5,9 +5,6 @@ plugins {
     id 'org.opensearch.migrations.java-application-conventions'
 }
 
-java.sourceCompatibility = JavaVersion.VERSION_11
-java.targetCompatibility = JavaVersion.VERSION_11
-
 dependencies {
     implementation project(':coreUtilities')
     implementation project(":RFS")

DEVELOPER_GUIDE.md

@@ -11,7 +11,7 @@
 
 ## Prerequisites
 
-- Java Development Kit (JDK) 11
+- Java Development Kit (JDK) 11-17
 - Gradle 8
 - Python3
 - Docker and Docker Compose (for local deployment)

DataGenerator/build.gradle

@@ -4,9 +4,6 @@ plugins {
     id 'io.freefair.lombok'
 }
 
-java.sourceCompatibility = JavaVersion.VERSION_11
-java.targetCompatibility = JavaVersion.VERSION_11
-
 dependencies {
     implementation project(":coreUtilities")
     implementation project(":RFS")

DocumentsFromSnapshotMigration/build.gradle

@@ -10,9 +10,6 @@ import com.bmuschko.gradle.docker.tasks.image.DockerBuildImage
 import groovy.transform.Canonical
 import org.opensearch.migrations.common.CommonUtils
 
-java.sourceCompatibility = JavaVersion.VERSION_11
-java.targetCompatibility = JavaVersion.VERSION_11
-
 @Canonical
 class DockerServiceProps {
     String projectName = ""

DocumentsFromSnapshotMigration/docker/Dockerfile

@@ -1,5 +1,5 @@
 # Using same base image as other Java containers in this repo
-FROM amazoncorretto:11-al2023-headless
+FROM amazoncorretto:17-al2023-headless
 
 # Install the AWS CLI in the container
 RUN dnf install -y aws-cli --setopt=install_weak_deps=False && \

DocumentsFromSnapshotMigration/src/main/java/org/opensearch/migrations/RfsMigrateDocuments.java

@@ -6,7 +6,9 @@
 import java.time.Clock;
 import java.time.Duration;
 import java.util.List;
+import java.util.Optional;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import org.opensearch.migrations.bulkload.common.DefaultSourceRepoAccessor;
 import org.opensearch.migrations.bulkload.common.DocumentReindexer;
@@ -259,16 +261,11 @@ public static void main(String[] args) throws Exception {
         var connectionContext = arguments.targetArgs.toConnectionContext();
 
 
-        String docTransformerConfig = TransformerConfigUtils.getTransformerConfig(arguments.docTransformationParams);
-        if (docTransformerConfig != null) {
-            log.atInfo().setMessage("Doc Transformations config string: {}")
-                    .addArgument(docTransformerConfig).log();
-        } else {
-            log.atInfo().setMessage("Using default transformation config: {}")
-                    .addArgument(DEFAULT_DOCUMENT_TRANSFORMATION_CONFIG).log();
-            docTransformerConfig = DEFAULT_DOCUMENT_TRANSFORMATION_CONFIG;
-        }
-        IJsonTransformer docTransformer = new TransformationLoader().getTransformerFactoryLoader(docTransformerConfig);
+        var docTransformerConfig = Optional.ofNullable(TransformerConfigUtils.getTransformerConfig(arguments.docTransformationParams))
+            .orElse(DEFAULT_DOCUMENT_TRANSFORMATION_CONFIG);
+        log.atInfo().setMessage("Doc Transformations config string: {}")
+                .addArgument(docTransformerConfig).log();
+        Supplier<IJsonTransformer> docTransformerSupplier = () -> new TransformationLoader().getTransformerFactoryLoader(docTransformerConfig);
 
         try (var processManager = new LeaseExpireTrigger(RfsMigrateDocuments::exitOnLeaseTimeout, Clock.systemUTC());
              var workCoordinator = new OpenSearchWorkCoordinator(
@@ -282,7 +279,7 @@ public static void main(String[] args) throws Exception {
                 arguments.numDocsPerBulkRequest,
                 arguments.numBytesPerBulkRequest,
                 arguments.maxConnections,
-                docTransformer);
+                docTransformerSupplier);
 
             SourceRepo sourceRepo;
             if (snapshotLocalDirPath == null) {

DocumentsFromSnapshotMigration/src/test/java/org/opensearch/migrations/bulkload/PerformanceVerificationTest.java

@@ -101,7 +101,7 @@ protected RfsLuceneDocument getDocument(IndexReader reader, int docId, boolean i
         int maxDocsPerBulkRequest = 1000;
         long maxBytesPerBulkRequest = Long.MAX_VALUE; // No Limit on Size
         int maxConcurrentWorkItems = 10;
-        DocumentReindexer reindexer = new DocumentReindexer(mockClient, maxDocsPerBulkRequest, maxBytesPerBulkRequest, maxConcurrentWorkItems, null);
+        DocumentReindexer reindexer = new DocumentReindexer(mockClient, maxDocsPerBulkRequest, maxBytesPerBulkRequest, maxConcurrentWorkItems, () -> null);
 
         // Create a mock IDocumentReindexContext
         IDocumentMigrationContexts.IDocumentReindexContext mockContext = mock(IDocumentMigrationContexts.IDocumentReindexContext.class);

DocumentsFromSnapshotMigration/src/test/java/org/opensearch/migrations/bulkload/SourceTestBase.java

@@ -280,7 +280,7 @@ public static DocumentsRunner.CompletionStatus migrateDocumentsWithOneWorker(
                         .host(targetAddress)
                         .compressionEnabled(compressionEnabled)
                         .build()
-                        .toConnectionContext()), 1000, Long.MAX_VALUE, 1, defaultDocTransformer),
+                        .toConnectionContext()), 1000, Long.MAX_VALUE, 1, () -> defaultDocTransformer),
                     new OpenSearchWorkCoordinator(
                         new CoordinateWorkHttpClient(ConnectionContextTestParams.builder()
                             .host(targetAddress)

MetadataMigration/build.gradle

@@ -4,9 +4,6 @@ plugins {
     id 'io.freefair.lombok'
 }
 
-java.sourceCompatibility = JavaVersion.VERSION_11
-java.targetCompatibility = JavaVersion.VERSION_11
-
 dependencies {
     implementation project(":coreUtilities")
     implementation project(":RFS")

RFS/build.gradle

@@ -6,9 +6,6 @@ plugins {
     id 'me.champeau.jmh'
 }
 
-java.sourceCompatibility = JavaVersion.VERSION_11
-java.targetCompatibility = JavaVersion.VERSION_11
-
 ext {
     awsSdkVersion = '2.25.16'
     dataset = findProperty('dataset') ?: 'skip_dataset'

RFS/src/main/java/org/opensearch/migrations/bulkload/common/DocumentReindexer.java

@@ -3,6 +3,7 @@
 import java.util.List;
 import java.util.UUID;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 
 import org.opensearch.migrations.reindexer.tracing.IDocumentMigrationContexts.IDocumentReindexContext;
 import org.opensearch.migrations.transform.IJsonTransformer;
@@ -23,13 +24,15 @@ public class DocumentReindexer {
     private final int maxDocsPerBulkRequest;
     private final long maxBytesPerBulkRequest;
     private final int maxConcurrentWorkItems;
-    private final IJsonTransformer transformer;
+    private final Supplier<IJsonTransformer> transformerSupplier;
 
     public Mono<Void> reindex(String indexName, Flux<RfsLuceneDocument> documentStream, IDocumentReindexContext context) {
+        // Transformers cannot be used simultaneously
+        var threadSafeTransformer = ThreadLocal.withInitial(transformerSupplier);
         var scheduler = Schedulers.newParallel("DocumentBulkAggregator");
         var bulkDocs = documentStream
             .publishOn(scheduler, 1)
-            .map(doc -> transformDocument(doc,indexName));
+            .map(doc -> transformDocument(threadSafeTransformer, doc,indexName));
 
         return this.reindexDocsInParallelBatches(bulkDocs, indexName, context)
             .doOnSuccess(unused -> log.debug("All batches processed"))
@@ -53,7 +56,8 @@ Mono<Void> reindexDocsInParallelBatches(Flux<BulkDocSection> docs, String indexN
     }
 
     @SneakyThrows
-    BulkDocSection transformDocument(RfsLuceneDocument doc, String indexName) {
+    BulkDocSection transformDocument(ThreadLocal<IJsonTransformer> transformerLocal, RfsLuceneDocument doc, String indexName) {
+        var transformer = transformerLocal.get();
         var original = new BulkDocSection(doc.id, indexName, doc.type, doc.source, doc.routing);
         if (transformer != null) {
             final Object transformedDoc = transformer.transformJson(original.toMap());

RFS/src/test/java/org/opensearch/migrations/bulkload/common/DocumentReindexerTest.java

@@ -50,7 +50,7 @@ class DocumentReindexerTest {
     @BeforeEach
     void setUp() {
         MockitoAnnotations.openMocks(this);
-        documentReindexer = new DocumentReindexer(mockClient, MAX_DOCS_PER_BULK, MAX_BYTES_PER_BULK_REQUEST, MAX_CONCURRENT_REQUESTS, null);
+        documentReindexer = new DocumentReindexer(mockClient, MAX_DOCS_PER_BULK, MAX_BYTES_PER_BULK_REQUEST, MAX_CONCURRENT_REQUESTS, () -> null);
         when(mockContext.createBulkRequest()).thenReturn(mock(IRfsContexts.IRequestContext.class));
     }
 
@@ -132,7 +132,7 @@ void reindex_shouldBufferByTransformedSize() throws JsonProcessingException {
 
         // Initialize DocumentReindexer with the transformer
         documentReindexer = new DocumentReindexer(
-            mockClient, numDocs, MAX_BYTES_PER_BULK_REQUEST, MAX_CONCURRENT_REQUESTS, transformer
+            mockClient, numDocs, MAX_BYTES_PER_BULK_REQUEST, MAX_CONCURRENT_REQUESTS, () -> transformer
         );
 
         Flux<RfsLuceneDocument> documentStream = Flux.range(1, numDocs)
@@ -237,7 +237,7 @@ private RfsLuceneDocument createLargeTestDocument(String id, int size) {
     void reindex_shouldRespectMaxConcurrentRequests() {
         int numDocs = 100;
         int maxConcurrentRequests = 5;
-        DocumentReindexer concurrentReindexer = new DocumentReindexer(mockClient, 1, MAX_BYTES_PER_BULK_REQUEST, maxConcurrentRequests, null);
+        DocumentReindexer concurrentReindexer = new DocumentReindexer(mockClient, 1, MAX_BYTES_PER_BULK_REQUEST, maxConcurrentRequests, () -> null);
 
         Flux<RfsLuceneDocument> documentStream = Flux.range(1, numDocs).map(i -> createTestDocument(String.valueOf(i)));
 
@@ -275,7 +275,7 @@ void reindex_shouldTransformDocuments() {
         IJsonTransformer transformer = new TransformationLoader().getTransformerFactoryLoader(CONFIG);
 
         // Initialize DocumentReindexer with the transformer
-        documentReindexer = new DocumentReindexer(mockClient, MAX_DOCS_PER_BULK, MAX_BYTES_PER_BULK_REQUEST, MAX_CONCURRENT_REQUESTS, transformer);
+        documentReindexer = new DocumentReindexer(mockClient, MAX_DOCS_PER_BULK, MAX_BYTES_PER_BULK_REQUEST, MAX_CONCURRENT_REQUESTS, () -> transformer);
 
         // Create a stream of documents, some requiring transformation and some not
         Flux<RfsLuceneDocument> documentStream = Flux.just(

RFS/src/test/java/org/opensearch/migrations/bulkload/framework/SimpleRestoreFromSnapshot_ES_7_10.java

@@ -82,7 +82,7 @@ public void updateTargetCluster(
                 ).readDocuments();
 
                 final var finalShardId = shardId;
-                new DocumentReindexer(client, 100, Long.MAX_VALUE, 1, null).reindex(index.getName(), documents, context)
+                new DocumentReindexer(client, 100, Long.MAX_VALUE, 1, () -> null).reindex(index.getName(), documents, context)
                     .doOnError(error -> logger.error("Error during reindexing: " + error))
                     .doOnSuccess(
                         done -> logger.info(

TrafficCapture/build.gradle

@@ -9,10 +9,25 @@ subprojects {
     // TODO: Expand to do more static checking in more projects
     if (project.name == "trafficReplayer" || project.name == "trafficCaptureProxyServer") {
         dependencies {
-            annotationProcessor group: 'com.google.errorprone', name: 'error_prone_core', version: '2.26.1'
+            annotationProcessor group: 'com.google.errorprone', name: 'error_prone_core', version: '2.32.0'
         }
         tasks.named('compileJava', JavaCompile) {
             if (project.name == "trafficReplayer" || project.name == "trafficCaptureProxyServer") {
+                options.fork = true
+                // Taken from https://errorprone.info/docs/installation
+                options.forkOptions.jvmArgs += [
+                    '--add-exports', 'jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED',
+                    '--add-exports', 'jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED',
+                    '--add-exports', 'jdk.compiler/com.sun.tools.javac.main=ALL-UNNAMED',
+                    '--add-exports', 'jdk.compiler/com.sun.tools.javac.model=ALL-UNNAMED',
+                    '--add-exports', 'jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED',
+                    '--add-exports', 'jdk.compiler/com.sun.tools.javac.processing=ALL-UNNAMED',
+                    '--add-exports', 'jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED',
+                    '--add-exports', 'jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED',
+                    '--add-opens', 'jdk.compiler/com.sun.tools.javac.code=ALL-UNNAMED',
+                    '--add-opens', 'jdk.compiler/com.sun.tools.javac.comp=ALL-UNNAMED',
+                ]
+
                 options.compilerArgs += [
                         "-XDcompilePolicy=simple",
                         "-Xplugin:ErrorProne -XepDisableAllChecks -Xep:MustBeClosed:ERROR -XepDisableWarningsInGeneratedCode",
@@ -25,12 +40,6 @@ subprojects {
 allprojects {
     apply plugin: 'java'
     apply plugin: 'org.owasp.dependencycheck'
-
-    java {
-        toolchain {
-            languageVersion = JavaLanguageVersion.of(11)
-        }
-    }
 }
 
 subprojects {

TrafficCapture/dockerSolution/README.md

@@ -91,7 +91,8 @@ within the final yaml configuration that is being output.
 
 ## Compatibility
 
-The tools in this directory can only be built if you have Java version 11 installed.
+The tools in this directory can only be built if you have Java version 11-17
+installed.
 
 The version is specified in `TrafficCapture/build.gradle` using a Java toolchain, which allows us
 to decouple the Java version used by Gradle itself from Java version used by the tools here.

TrafficCapture/trafficCaptureProxyServer/build.gradle

@@ -65,3 +65,94 @@ application {
     // Define the main class for the application.
     mainClass = 'org.opensearch.migrations.trafficcapture.proxyserver.CaptureProxy'
 }
+
+tasks.register("verifyCompatibilityWithJDK11") {
+    group = "verification"
+    description = "Verify Java major versions of compiled classes and dependencies for compatibility with JDK 11."
+
+    // Declare task inputs
+    inputs.files(fileTree("${buildDir}/classes/java/main")).withPropertyName("compiledClasses")
+    inputs.files(configurations.compileClasspath).withPropertyName("compileClasspath")
+    inputs.files(configurations.runtimeClasspath).withPropertyName("runtimeClasspath")
+
+    // Declare outputs (optional if no files are written, just up-to-date checking)
+    outputs.upToDateWhen { true }
+
+    doLast {
+        // Task implementation logic
+        def isCompatible = true
+        def analyzedFiles = [] // To avoid duplicate processing
+
+        def getMajorVersion = { InputStream inputStream ->
+            inputStream.skip(6)
+            def majorVersion = inputStream.read() << 8 | inputStream.read()
+            inputStream.close()
+            return majorVersion
+        }
+
+        def processClassFile = { InputStream inputStream, String name, String source ->
+            def majorVersion = getMajorVersion(inputStream)
+            if (majorVersion > 55) {
+                isCompatible = false
+                println "Incompatible class file in ${source}: ${name} (Java major version ${majorVersion})"
+            }
+        }
+
+        def analyzeCompiledClasses = {
+            def outputDir = file("${buildDir}/classes/java/main")
+            if (outputDir.exists()) {
+                println "Analyzing compiled classes in ${outputDir}..."
+                fileTree(dir: outputDir, include: '**/*.class').forEach { classFile ->
+                    classFile.withInputStream { inputStream ->
+                        processClassFile(inputStream, classFile.name, outputDir.toString())
+                    }
+                }
+            } else {
+                println "No compiled classes found in ${outputDir}."
+            }
+        }
+
+        def analyzeDependencies = { configuration ->
+            println "\nAnalyzing ${configuration.name} dependencies..."
+            configuration.files.each { dependency ->
+                if (dependency.name.endsWith(".jar") && !analyzedFiles.contains(dependency)) {
+                    analyzedFiles.add(dependency)
+                    def zipStream = new java.util.zip.ZipInputStream(new FileInputStream(dependency))
+                    def entry
+                    while ((entry = zipStream.nextEntry) != null) {
+                        if (entry.name.endsWith(".class")) {
+                            if (entry.name.startsWith("META-INF/versions/")) {
+                                def versionDir = entry.name.split("/")[2]
+                                def versionNumber = versionDir.isInteger() ? versionDir.toInteger() : 0
+                                if (versionNumber > 11) {
+                                    continue
+                                }
+                            }
+                            def tempFile = File.createTempFile("class", ".tmp")
+                            tempFile.deleteOnExit()
+                            tempFile.withOutputStream { it << zipStream }
+                            tempFile.withInputStream { inputStream ->
+                                processClassFile(inputStream, entry.name, dependency.name)
+                            }
+                        }
+                    }
+                    zipStream.close()
+                }
+            }
+        }
+
+        analyzeCompiledClasses()
+        analyzeDependencies(configurations.compileClasspath)
+        analyzeDependencies(configurations.runtimeClasspath)
+
+        if (!isCompatible) {
+            throw new GradleException("Incompatible class files detected! Ensure all classes are compatible with JDK 11 (major version ≤ 55).")
+        }
+    }
+    dependsOn tasks.named("assemble")
+}
+
+// Add class version verification to slowTest
+tasks.named("slowTest") {
+    finalizedBy("verifyCompatibilityWithJDK11")
+}