Add 6.8 to lease expiration tests

Signed-off-by: Mikayla Thompson <thomika@amazon.com>
mikaylathompson · Jan 17, 2025 · 021322c · 021322c
1 parent c6a98d4
commit 021322c
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 17 deletions.
diff --git a/...apshotMigration/src/test/java/org/opensearch/migrations/bulkload/LeaseExpirationTest.java b/...apshotMigration/src/test/java/org/opensearch/migrations/bulkload/LeaseExpirationTest.java
@@ -6,8 +6,13 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import java.util.stream.Stream;
 
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.opensearch.migrations.CreateSnapshot;
+import org.opensearch.migrations.DataGenerator;
+import org.opensearch.migrations.DataGeneratorArgs;
 import org.opensearch.migrations.bulkload.common.OpenSearchClientFactory;
 import org.opensearch.migrations.bulkload.common.http.ConnectionContextTestParams;
 import org.opensearch.migrations.bulkload.framework.SearchClusterContainer;
@@ -35,9 +40,18 @@ public class LeaseExpirationTest extends SourceTestBase {
 
     public static final String TARGET_DOCKER_HOSTNAME = "target";
 
-    @ParameterizedTest
-    @ValueSource(booleans = {true, false})
-    public void testProcessExitsAsExpected(boolean forceMoreSegments) {
+    private static Stream<Arguments> testParameters() {
+        List<Boolean> forceMoreSegmentsValues = List.of(false, true);
+        List<SearchClusterContainer.ContainerVersion> sourceClusterVersions = List.of(SearchClusterContainer.ES_V6_8_23, SearchClusterContainer.ES_V7_10_2);
+
+        return forceMoreSegmentsValues.stream()
+                .flatMap(force -> sourceClusterVersions.stream()
+                        .map(version -> Arguments.of(force, version)));
+    }
+
+    @ParameterizedTest(name = "forceMoreSegments={0}, sourceClusterVersion={1}")
+    @MethodSource("testParameters")
+    public void testProcessExitsAsExpected(boolean forceMoreSegments, SearchClusterContainer.ContainerVersion sourceClusterVersion) {
         // Sending 5 docs per request with 4 requests concurrently with each taking 0.250 second is 80 docs/sec
         // will process 9680 docs in 121 seconds. With 40s lease duration, expect to be finished in 4 leases.
         // This is ensured with the toxiproxy settings, the migration should not be able to be completed
@@ -50,24 +64,24 @@ public void testProcessExitsAsExpected(boolean forceMoreSegments) {
         int continueExitCode = 2;
         int finalExitCodePerShard = 0;
         runTestProcessWithCheckpoint(continueExitCode, (migrationProcessesPerShard - 1) * shards,
-                finalExitCodePerShard, shards, shards, indexDocCount, forceMoreSegments,
-            d -> runProcessAgainstToxicTarget(d.tempDirSnapshot, d.tempDirLucene, d.proxyContainer
+                finalExitCodePerShard, shards, shards, indexDocCount, forceMoreSegments, sourceClusterVersion,
+            d -> runProcessAgainstToxicTarget(d.tempDirSnapshot, d.tempDirLucene, d.proxyContainer, sourceClusterVersion
             ));
     }
 
     @SneakyThrows
     private void runTestProcessWithCheckpoint(int expectedInitialExitCode, int expectedInitialExitCodeCount,
                                               int expectedEventualExitCode, int expectedEventualExitCodeCount,
                                               int shards, int indexDocCount,
-                                              boolean forceMoreSegments,
+                                              boolean forceMoreSegments, SearchClusterContainer.ContainerVersion sourceClusterVersion,
                                               Function<RunData, Integer> processRunner) {
         final var testSnapshotContext = SnapshotTestContext.factory().noOtelTracking();
 
         var tempDirSnapshot = Files.createTempDirectory("opensearchMigrationReindexFromSnapshot_test_snapshot");
         var tempDirLucene = Files.createTempDirectory("opensearchMigrationReindexFromSnapshot_test_lucene");
 
         try (
-            var esSourceContainer = new SearchClusterContainer(SearchClusterContainer.ES_V7_10_2)
+            var esSourceContainer = new SearchClusterContainer(sourceClusterVersion)
                     .withAccessToHost(true);
             var network = Network.newNetwork();
             var osTargetContainer = new SearchClusterContainer(SearchClusterContainer.OS_V2_14_0)
@@ -84,14 +98,15 @@ private void runTestProcessWithCheckpoint(int expectedInitialExitCode, int expec
             proxyContainer.start("target", 9200);
 
             // Populate the source cluster with data
-            var client = new OpenSearchClientFactory(ConnectionContextTestParams.builder()
-                .host(esSourceContainer.getUrl())
-                .build()
-                .toConnectionContext()
-            ).determineVersionAndCreate();
+            var clientFactory = new OpenSearchClientFactory(ConnectionContextTestParams.builder()
+                    .host(esSourceContainer.getUrl())
+                    .build()
+                    .toConnectionContext());
+            var client = clientFactory.determineVersionAndCreate();
             var generator = new WorkloadGenerator(client);
             var workloadOptions = new WorkloadOptions();
 
+
             var sourceClusterOperations = new ClusterOperations(esSourceContainer.getUrl());
 
             // Number of default shards is different across different versions on ES/OS.
@@ -177,7 +192,8 @@ private void runTestProcessWithCheckpoint(int expectedInitialExitCode, int expec
     private static int runProcessAgainstToxicTarget(
         Path tempDirSnapshot,
         Path tempDirLucene,
-        ToxiProxyWrapper proxyContainer
+        ToxiProxyWrapper proxyContainer,
+        SearchClusterContainer.ContainerVersion sourceClusterVersion
     ) {
         String targetAddress = proxyContainer.getProxyUriAsString();
         var tp = proxyContainer.getProxy();
@@ -189,7 +205,8 @@ private static int runProcessAgainstToxicTarget(
         String[] additionalArgs = {
             "--documents-per-bulk-request", "5",
             "--max-connections", "4",
-            "--initial-lease-duration", "PT40s"
+            "--initial-lease-duration", "PT40s",
+            "--source-version", sourceClusterVersion.getVersion().toString()
         };
 
         ProcessBuilder processBuilder = setupProcess(

diff --git a/...romSnapshotMigration/src/test/java/org/opensearch/migrations/bulkload/SourceTestBase.java b/...romSnapshotMigration/src/test/java/org/opensearch/migrations/bulkload/SourceTestBase.java
@@ -358,15 +358,17 @@ public static ProcessBuilder setupProcess(
             "--target-host",
             targetAddress,
             "--index-allowlist",
-            "geonames",
-            "--source-version",
-            "ES_7_10"
+            "geonames"
         ));
 
         if (additionalArgs != null && additionalArgs.length > 0) {
             argsList.addAll(Arrays.asList(additionalArgs));
+            if (!argsList.contains("--source-version")) {
+                argsList.addAll(Arrays.asList("--source-version", "ES_7_10"));
+            }
         }
 
+
         log.atInfo().setMessage("Running RfsMigrateDocuments with args: {}")
             .addArgument(() -> argsList.toString())
             .log();