[Transform] re-enable BWC and cleanup after backport (#53012)

re-enable BWC tests and adapt code on master for BWC, remove code
no longer necessary

relates #52998
relates #53010

Author: Hendrik Muhs

build.gradle

@@ -219,8 +219,8 @@ task verifyVersions {
  * after the backport of the backcompat code is complete.
  */
 
-boolean bwc_tests_enabled = false
-final String bwc_tests_disabled_issue = "https://github.com/elastic/elasticsearch/pull/52998" /* place a PR link here when committing bwc changes */
+boolean bwc_tests_enabled = true
+final String bwc_tests_disabled_issue = "" /* place a PR link here when committing bwc changes */
 if (bwc_tests_enabled == false) {
   if (bwc_tests_disabled_issue.isEmpty()) {
     throw new GradleException("bwc_tests_disabled_issue must be set when bwc_tests_enabled == false")

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/TransformTaskParams.java

@@ -68,7 +68,7 @@ public TransformTaskParams(StreamInput in) throws IOException {
         } else {
             this.frequency = null;
         }
-        if (in.getVersion().onOrAfter(Version.V_8_0_0)) { // todo: V_7_7_0
+        if (in.getVersion().onOrAfter(Version.V_7_7_0)) {
             this.requiresRemote = in.readBoolean();
         } else {
             this.requiresRemote = false;
@@ -94,7 +94,7 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_7_3_0)) {
             out.writeOptionalTimeValue(frequency);
         }
-        if (out.getVersion().onOrAfter(Version.V_8_0_0)) { // todo: V_7_7_0
+        if (out.getVersion().onOrAfter(Version.V_7_7_0)) {
             out.writeBoolean(requiresRemote);
         }
     }

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformPersistentTasksExecutor.java

@@ -11,7 +11,6 @@
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ResourceNotFoundException;
-import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.LatchedActionListener;
 import org.elasticsearch.action.support.IndicesOptions;
@@ -103,19 +102,13 @@ public PersistentTasksCustomMetaData.Assignment getAssignment(TransformTaskParam
         }
         DiscoveryNode discoveryNode = selectLeastLoadedNode(
             clusterState,
-            (node) -> node.getVersion().onOrAfter(Version.V_8_0_0)
-                ? nodeCanRunThisTransform(node, params, null)
-                : nodeCanRunThisTransformPre77(node, params, null)
+            (node) -> nodeCanRunThisTransform(node, params, null)
         );
 
         if (discoveryNode == null) {
             Map<String, String> explainWhyAssignmentFailed = new TreeMap<>();
             for (DiscoveryNode node : clusterState.getNodes()) {
-                if (node.getVersion().onOrAfter(Version.V_8_0_0)) { // todo: V_7_7_0, remove from 8.0
-                    nodeCanRunThisTransform(node, params, explainWhyAssignmentFailed);
-                } else {
-                    nodeCanRunThisTransformPre77(node, params, explainWhyAssignmentFailed);
-                }
+                nodeCanRunThisTransform(node, params, explainWhyAssignmentFailed);
             }
             String reason = "Not starting transform ["
                 + params.getId()
@@ -130,29 +123,6 @@ public PersistentTasksCustomMetaData.Assignment getAssignment(TransformTaskParam
         return new PersistentTasksCustomMetaData.Assignment(discoveryNode.getId(), "");
     }
 
-    // todo: this can be removed for 8.0 after backport
-    public static boolean nodeCanRunThisTransformPre77(DiscoveryNode node, TransformTaskParams params, Map<String, String> explain) {
-        if (node.isDataNode() == false) {
-            if (explain != null) {
-                explain.put(node.getId(), "not a data node");
-            }
-            return false;
-        }
-
-        // version of the transform run on a node that has at least the same version
-        if (node.getVersion().onOrAfter(params.getVersion()) == false) {
-            if (explain != null) {
-                explain.put(
-                    node.getId(),
-                    "node has version: " + node.getVersion() + " but transform requires at least " + params.getVersion()
-                );
-            }
-            return false;
-        }
-
-        return true;
-    }
-
     public static boolean nodeCanRunThisTransform(DiscoveryNode node, TransformTaskParams params, Map<String, String> explain) {
         // version of the transform run on a node that has at least the same version
         if (node.getVersion().onOrAfter(params.getVersion()) == false) {

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/transforms/TransformPersistentTasksExecutorTests.java

@@ -70,7 +70,7 @@ public void testNodeVersionAssignment() {
             equalTo("current-data-node-with-0-tasks-transform-remote-disabled")
         );
         assertThat(
-            executor.getAssignment(new TransformTaskParams("new-old-task-id", Version.V_7_5_0, null, true), cs).getExecutorNode(),
+            executor.getAssignment(new TransformTaskParams("new-old-task-id", Version.V_7_7_0, null, true), cs).getExecutorNode(),
             equalTo("past-data-node-1")
         );
     }
@@ -102,15 +102,15 @@ public void testNodeAssignmentProblems() {
         cs = buildClusterState(nodes);
         executor = buildTaskExecutor();
 
-        assignment = executor.getAssignment(new TransformTaskParams("new-task-id", Version.V_7_7_0, null, false), cs);
+        assignment = executor.getAssignment(new TransformTaskParams("new-task-id", Version.V_8_0_0, null, false), cs);
         assertNull(assignment.getExecutorNode());
         assertThat(
             assignment.getExplanation(),
             equalTo(
                 "Not starting transform [new-task-id], reasons ["
                     + "current-data-node-with-transform-disabled:not a transform node"
                     + "|"
-                    + "past-data-node-1:node has version: 7.5.0 but transform requires at least 7.7.0"
+                    + "past-data-node-1:node has version: 7.7.0 but transform requires at least 8.0.0"
                     + "]"
             )
         );
@@ -276,9 +276,9 @@ private DiscoveryNodes.Builder buildNodes(
                 new DiscoveryNode(
                     "past-data-node-1",
                     buildNewFakeTransportAddress(),
-                    Collections.emptyMap(),
+                    transformNodeAttributes,
                     new HashSet<>(Arrays.asList(DiscoveryNodeRole.DATA_ROLE, DiscoveryNodeRole.MASTER_ROLE)),
-                    Version.V_7_5_0
+                    Version.V_7_7_0
                 )
             );
         }