Replace Version with RemoteVersion in reindex module (opensearch-project#18457)

- Add RemoteVersion class with backward compatibility for Elasticsearch versions
- Support version parsing from string with distribution parameter
- Include version comparison methods (before, after, onOrAfter, onOrBefore)
- Support both Elasticsearch and OpenSearch version constants

---------

Signed-off-by: Hyunsang Han <[email protected]>

Author: HyunSangHan

modules/reindex/src/main/java/org/opensearch/index/reindex/remote/RemoteRequestBuilders.java

@@ -35,7 +35,6 @@
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.io.entity.StringEntity;
 import org.opensearch.OpenSearchException;
-import org.opensearch.Version;
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.client.Request;
 import org.opensearch.common.logging.DeprecationLogger;
@@ -74,7 +73,7 @@ final class RemoteRequestBuilders {
 
     private RemoteRequestBuilders() {}
 
-    static Request initialSearch(SearchRequest searchRequest, BytesReference query, Version remoteVersion) {
+    static Request initialSearch(SearchRequest searchRequest, BytesReference query, RemoteVersion remoteVersion) {
         // It is nasty to build paths with StringBuilder but we'll be careful....
         StringBuilder path = new StringBuilder("/");
         addIndices(path, searchRequest.indices());
@@ -84,7 +83,7 @@ static Request initialSearch(SearchRequest searchRequest, BytesReference query,
         if (searchRequest.scroll() != null) {
             TimeValue keepAlive = searchRequest.scroll().keepAlive();
             // V_5_0_0
-            if (remoteVersion.before(Version.fromId(5000099))) {
+            if (remoteVersion.before(RemoteVersion.ELASTICSEARCH_5_0_0)) {
                 /* Versions of Elasticsearch before 5.0 couldn't parse nanos or micros
                  * so we toss out that resolution, rounding up because more scroll
                  * timeout seems safer than less. */
@@ -103,7 +102,7 @@ static Request initialSearch(SearchRequest searchRequest, BytesReference query,
         if (searchRequest.source().sorts() != null) {
             boolean useScan = false;
             // Detect if we should use search_type=scan rather than a sort
-            if (remoteVersion.before(Version.fromId(2010099))) {
+            if (remoteVersion.before(RemoteVersion.ELASTICSEARCH_2_1_0)) {
                 for (SortBuilder<?> sort : searchRequest.source().sorts()) {
                     if (sort instanceof FieldSortBuilder) {
                         FieldSortBuilder f = (FieldSortBuilder) sort;
@@ -124,10 +123,10 @@ static Request initialSearch(SearchRequest searchRequest, BytesReference query,
                 request.addParameter("sort", sorts.toString());
             }
         }
-        if (remoteVersion.before(Version.fromId(2000099))) {
+        if (remoteVersion.before(RemoteVersion.ELASTICSEARCH_2_0_0)) {
             // Versions before 2.0.0 need prompting to return interesting fields. Note that timestamp isn't available at all....
             searchRequest.source().storedField("_parent").storedField("_routing").storedField("_ttl");
-            if (remoteVersion.before(Version.fromId(1000099))) {
+            if (remoteVersion.before(RemoteVersion.ELASTICSEARCH_1_0_0)) {
                 // Versions before 1.0.0 don't support `"_source": true` so we have to ask for the _source in a funny way.
                 if (false == searchRequest.source().storedFields().fieldNames().contains("_source")) {
                     searchRequest.source().storedField("_source");
@@ -140,11 +139,11 @@ static Request initialSearch(SearchRequest searchRequest, BytesReference query,
                 fields.append(',').append(searchRequest.source().storedFields().fieldNames().get(i));
             }
             // V_5_0_0
-            String storedFieldsParamName = remoteVersion.before(Version.fromId(5000099)) ? "fields" : "stored_fields";
+            String storedFieldsParamName = remoteVersion.before(RemoteVersion.ELASTICSEARCH_5_0_0) ? "fields" : "stored_fields";
             request.addParameter(storedFieldsParamName, fields.toString());
         }
 
-        if (remoteVersion.onOrAfter(Version.fromId(6030099))) {
+        if (remoteVersion.onOrAfter(RemoteVersion.ELASTICSEARCH_6_3_0)) {
             // allow_partial_results introduced in 6.3, running remote reindex against earlier versions still silently discards RED shards.
             request.addParameter("allow_partial_search_results", "false");
         }
@@ -173,7 +172,7 @@ static Request initialSearch(SearchRequest searchRequest, BytesReference query,
             if (searchRequest.source().fetchSource() != null) {
                 entity.field("_source", searchRequest.source().fetchSource());
             } else {
-                if (remoteVersion.onOrAfter(Version.fromId(1000099))) {
+                if (remoteVersion.onOrAfter(RemoteVersion.ELASTICSEARCH_1_0_0)) {
                     // Versions before 1.0 don't support `"_source": true` so we have to ask for the source as a stored field.
                     entity.field("_source", true);
                 }
@@ -225,19 +224,18 @@ private static String sortToUri(SortBuilder<?> sort) {
         throw new IllegalArgumentException("Unsupported sort [" + sort + "]");
     }
 
-    static Request scroll(String scroll, TimeValue keepAlive, Version remoteVersion) {
+    static Request scroll(String scroll, TimeValue keepAlive, RemoteVersion remoteVersion) {
         Request request = new Request("POST", "/_search/scroll");
 
-        // V_5_0_0
-        if (remoteVersion.before(Version.fromId(5000099))) {
+        if (remoteVersion.before(RemoteVersion.ELASTICSEARCH_5_0_0)) {
             /* Versions of Elasticsearch before 5.0 couldn't parse nanos or micros
              * so we toss out that resolution, rounding up so we shouldn't end up
              * with 0s. */
             keepAlive = timeValueMillis((long) Math.ceil(keepAlive.millisFrac()));
         }
         request.addParameter("scroll", keepAlive.getStringRep());
 
-        if (remoteVersion.before(Version.fromId(2000099))) {
+        if (remoteVersion.before(RemoteVersion.ELASTICSEARCH_2_0_0)) {
             // Versions before 2.0.0 extract the plain scroll_id from the body
             request.setEntity(new StringEntity(scroll, ContentType.TEXT_PLAIN));
             return request;
@@ -252,10 +250,10 @@ static Request scroll(String scroll, TimeValue keepAlive, Version remoteVersion)
         return request;
     }
 
-    static Request clearScroll(String scroll, Version remoteVersion) {
+    static Request clearScroll(String scroll, RemoteVersion remoteVersion) {
         Request request = new Request("DELETE", "/_search/scroll");
 
-        if (remoteVersion.before(Version.fromId(2000099))) {
+        if (remoteVersion.before(RemoteVersion.ELASTICSEARCH_2_0_0)) {
             // Versions before 2.0.0 extract the plain scroll_id from the body
             request.setEntity(new StringEntity(scroll, ContentType.TEXT_PLAIN));
             return request;

modules/reindex/src/main/java/org/opensearch/index/reindex/remote/RemoteResponseParsers.java

@@ -33,8 +33,6 @@
 package org.opensearch.index.reindex.remote;
 
 import org.apache.lucene.search.TotalHits;
-import org.opensearch.LegacyESVersion;
-import org.opensearch.Version;
 import org.opensearch.common.collect.Tuple;
 import org.opensearch.core.ParseField;
 import org.opensearch.core.common.ParsingException;
@@ -294,20 +292,21 @@ public void setCausedBy(Throwable causedBy) {
     }
 
     /**
-     * Parses the main action to return just the {@linkplain Version} that it returns. We throw everything else out.
+     * Parses the main action to return just the {@linkplain RemoteVersion} that it returns. We throw everything else out.
      */
-    public static final ConstructingObjectParser<Version, MediaType> MAIN_ACTION_PARSER = new ConstructingObjectParser<>(
+    public static final ConstructingObjectParser<RemoteVersion, MediaType> MAIN_ACTION_PARSER = new ConstructingObjectParser<>(
         "/",
         true,
-        a -> (Version) a[0]
+        a -> (RemoteVersion) a[0]
     );
     static {
-        ConstructingObjectParser<Version, MediaType> versionParser = new ConstructingObjectParser<>(
+        ConstructingObjectParser<RemoteVersion, MediaType> versionParser = new ConstructingObjectParser<>(
             "version",
             true,
-            a -> a[0] == null
-                ? LegacyESVersion.fromString(((String) a[1]).replace("-SNAPSHOT", "").replaceFirst("-(alpha\\d+|beta\\d+|rc\\d+)", ""))
-                : Version.fromString(((String) a[1]).replace("-SNAPSHOT", "").replaceFirst("-(alpha\\d+|beta\\d+|rc\\d+)", ""))
+            a -> RemoteVersion.fromString(
+                ((String) a[1]).replace("-SNAPSHOT", "").replaceFirst("-(alpha\\d+|beta\\d+|rc\\d+)", ""),
+                a[0] != null
+            )
         );
         versionParser.declareStringOrNull(optionalConstructorArg(), new ParseField("distribution"));
         versionParser.declareString(constructorArg(), new ParseField("number"));

modules/reindex/src/main/java/org/opensearch/index/reindex/remote/RemoteScrollableHitSource.java

@@ -42,7 +42,6 @@
 import org.apache.logging.log4j.util.Supplier;
 import org.opensearch.OpenSearchException;
 import org.opensearch.OpenSearchStatusException;
-import org.opensearch.Version;
 import org.opensearch.action.bulk.BackoffPolicy;
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.client.Request;
@@ -81,7 +80,7 @@ public class RemoteScrollableHitSource extends ScrollableHitSource {
     private final RestClient client;
     private final BytesReference query;
     private final SearchRequest searchRequest;
-    Version remoteVersion;
+    RemoteVersion remoteVersion;
 
     public RemoteScrollableHitSource(
         Logger logger,
@@ -116,7 +115,7 @@ protected void doStart(RejectAwareActionListener<Response> searchListener) {
         }, searchListener::onFailure, searchListener::onFailure));
     }
 
-    void lookupRemoteVersion(RejectAwareActionListener<Version> listener) {
+    void lookupRemoteVersion(RejectAwareActionListener<RemoteVersion> listener) {
         logger.trace("Checking version for remote domain");
         // We're skipping retries for the first call to remote cluster so that we fail fast & respond back immediately
         // instead of retrying for longer duration.
@@ -159,7 +158,8 @@ public void onFailure(Exception e) {
             private void logFailure(Exception e) {
                 if (e instanceof ResponseException) {
                     ResponseException re = (ResponseException) e;
-                    if (remoteVersion.before(Version.fromId(2000099)) && re.getResponse().getStatusLine().getStatusCode() == 404) {
+                    if (remoteVersion.before(RemoteVersion.ELASTICSEARCH_2_0_0)
+                        && re.getResponse().getStatusLine().getStatusCode() == 404) {
                         logger.debug(
                             (Supplier<?>) () -> new ParameterizedMessage(
                                 "Failed to clear scroll [{}] from pre-2.0 OpenSearch. This is normal if the request terminated "

modules/reindex/src/main/java/org/opensearch/index/reindex/remote/RemoteVersion.java

@@ -0,0 +1,149 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.reindex.remote;
+
+import java.util.Objects;
+
+/**
+ * Represents a version of a remote Elasticsearch or OpenSearch cluster for reindexing purposes.
+ * This class provides backward compatibility support for communicating with older Elasticsearch versions
+ * without relying on the global Version class.
+ */
+public final class RemoteVersion implements Comparable<RemoteVersion> {
+
+    private final int major;
+    private final int minor;
+    private final int revision;
+    final boolean isOpenSearch;
+
+    // Common version constants for backward compatibility
+    public static final RemoteVersion ELASTICSEARCH_0_20_5 = new RemoteVersion(0, 20, 5, false);
+    public static final RemoteVersion ELASTICSEARCH_0_90_13 = new RemoteVersion(0, 90, 13, false);
+    public static final RemoteVersion ELASTICSEARCH_1_0_0 = new RemoteVersion(1, 0, 0, false);
+    public static final RemoteVersion ELASTICSEARCH_1_7_5 = new RemoteVersion(1, 7, 5, false);
+    public static final RemoteVersion ELASTICSEARCH_2_0_0 = new RemoteVersion(2, 0, 0, false);
+    public static final RemoteVersion ELASTICSEARCH_2_1_0 = new RemoteVersion(2, 1, 0, false);
+    public static final RemoteVersion ELASTICSEARCH_2_3_3 = new RemoteVersion(2, 3, 3, false);
+    public static final RemoteVersion ELASTICSEARCH_5_0_0 = new RemoteVersion(5, 0, 0, false);
+    public static final RemoteVersion ELASTICSEARCH_6_0_0 = new RemoteVersion(6, 0, 0, false);
+    public static final RemoteVersion ELASTICSEARCH_6_3_0 = new RemoteVersion(6, 3, 0, false);
+    public static final RemoteVersion ELASTICSEARCH_7_0_0 = new RemoteVersion(7, 0, 0, false);
+
+    // OpenSearch versions
+    public static final RemoteVersion OPENSEARCH_1_0_0 = new RemoteVersion(1, 0, 0, true);
+    public static final RemoteVersion OPENSEARCH_2_0_0 = new RemoteVersion(2, 0, 0, true);
+    public static final RemoteVersion OPENSEARCH_3_1_0 = new RemoteVersion(3, 1, 0, true);
+
+    public RemoteVersion(int major, int minor, int revision, boolean isOpenSearch) {
+        this.major = major;
+        this.minor = minor;
+        this.revision = revision;
+        this.isOpenSearch = isOpenSearch;
+    }
+
+    /**
+     * Parse version string like "7.10.2" or "1.0.0"
+     */
+    public static RemoteVersion fromString(String version, boolean isOpenSearch) {
+        if (version == null || version.trim().isEmpty()) {
+            throw new IllegalArgumentException("Version string cannot be null or empty");
+        }
+
+        // Remove snapshot and pre-release qualifiers
+        String cleanVersion = version.replace("-SNAPSHOT", "").replaceFirst("-(alpha\\d+|beta\\d+|rc\\d+)", "");
+
+        String[] parts = cleanVersion.split("\\.");
+        if (parts.length < 2) {
+            throw new IllegalArgumentException("Invalid version format: " + version);
+        }
+
+        try {
+            int major = Integer.parseInt(parts[0]);
+            int minor = Integer.parseInt(parts[1]);
+            int revision = parts.length > 2 ? Integer.parseInt(parts[2]) : 0;
+
+            return new RemoteVersion(major, minor, revision, isOpenSearch);
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException("Invalid version format: " + version, e);
+        }
+    }
+
+    public boolean before(RemoteVersion other) {
+        return this.compareTo(other) < 0;
+    }
+
+    public boolean onOrAfter(RemoteVersion other) {
+        return this.compareTo(other) >= 0;
+    }
+
+    public boolean onOrBefore(RemoteVersion other) {
+        return this.compareTo(other) <= 0;
+    }
+
+    public boolean after(RemoteVersion other) {
+        return this.compareTo(other) > 0;
+    }
+
+    @Override
+    public int compareTo(RemoteVersion other) {
+        if (other == null) {
+            return 1;
+        }
+
+        if (this.isOpenSearch != other.isOpenSearch) {
+            return this.isOpenSearch ? 1 : -1;
+        }
+
+        int result = Integer.compare(this.major, other.major);
+        if (result != 0) {
+            return result;
+        }
+
+        result = Integer.compare(this.minor, other.minor);
+        if (result != 0) {
+            return result;
+        }
+
+        return Integer.compare(this.revision, other.revision);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+        RemoteVersion that = (RemoteVersion) obj;
+        return major == that.major && minor == that.minor && revision == that.revision && Objects.equals(isOpenSearch, that.isOpenSearch);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(major, minor, revision, isOpenSearch);
+    }
+
+    @Override
+    public String toString() {
+        return major + "." + minor + "." + revision;
+    }
+
+    public int getMajor() {
+        return major;
+    }
+
+    public int getMinor() {
+        return minor;
+    }
+
+    public int getRevision() {
+        return revision;
+    }
+}