diff --git a/.github/workflows/test_bwc.yml b/.github/workflows/test_bwc.yml
new file mode 100644
index 0000000000..83c3ca7eb5
--- /dev/null
+++ b/.github/workflows/test_bwc.yml
@@ -0,0 +1,40 @@
+name: Test MLCommons BWC
+on:
+  push:
+    branches:
+      - "*"
+  pull_request:
+    branches:
+      - "*"
+
+jobs:
+  Build-ad:
+    strategy:
+      matrix:
+        java: [11,17]
+      fail-fast: false
+
+    name: Test MLCommons BWC
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Setup Java ${{ matrix.java }}
+        uses: actions/setup-java@v1
+        with:
+          java-version: ${{ matrix.java }}
+
+      # ml-commons
+      - name: Checkout MLCommons
+        uses: actions/checkout@v2
+
+      - name: Assemble MLCommons
+        run: |
+          plugin_version=`./gradlew properties -q | grep "opensearch_build:" | awk '{print $2}'`
+          echo plugin_version $plugin_version
+          ./gradlew assemble
+          echo "Creating ./plugin/src/test/resources/org/opensearch/ml/bwc..."
+          mkdir -p ./plugin/src/test/resources/org/opensearch/ml/bwc
+      - name: Run MLCommons Backwards Compatibility Tests
+        run: |
+          echo "Running backwards compatibility tests ..."
+          ./gradlew bwcTestSuite -Dtests.security.manager=false
\ No newline at end of file
diff --git a/.gitignore b/.gitignore
index e1c2d340ff..4269de1628 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,4 +5,5 @@ client/build/
 common/build/
 ml-algorithms/build/
 plugin/build/
+plugin/src/test/resources/bwc/*
 .DS_Store
diff --git a/DEVELOPER_GUIDE.md b/DEVELOPER_GUIDE.md
index dcb9d71f54..2d9f214f87 100644
--- a/DEVELOPER_GUIDE.md
+++ b/DEVELOPER_GUIDE.md
@@ -40,6 +40,10 @@ This package uses the [Gradle](https://docs.gradle.org/current/userguide/usergui
 6. `./gradlew integTest -Dtests.method="<method name>"` run specific integ test method, for example `./gradlew integTest -Dtests.method="testTrainAndPredictKmeans"`
 7. `./gradlew integTest -Dtests.rest.cluster=localhost:9200 -Dtests.cluster=localhost:9200 -Dtests.clustername="docker-cluster" -Dhttps=true -Duser=admin -Dpassword=admin` launches integration tests against a local cluster and run tests with security. Detail steps: (1)download OpenSearch tarball to local and install by running `opensearch-tar-install.sh`; (2)build ML plugin zip with your change and install ML plugin zip; (3)restart local test cluster; (4) run this gradle command to test.  
 8. `./gradlew spotlessApply` formats code. And/or import formatting rules in `.eclipseformat.xml` with IDE.
+9. `./gradlew adBwcCluster#mixedClusterTask -Dtests.security.manager=false` launches a cluster with three nodes of bwc version of OpenSearch with anomaly-detection and job-scheduler and tests backwards compatibility by upgrading one of the nodes with the current version of OpenSearch with anomaly-detection and job-scheduler creating a mixed cluster.
+10. `./gradlew adBwcCluster#rollingUpgradeClusterTask -Dtests.security.manager=false` launches a cluster with three nodes of bwc version of OpenSearch with anomaly-detection and job-scheduler and tests backwards compatibility by performing rolling upgrade of each node with the current version of OpenSearch with anomaly-detection and job-scheduler.
+11. `./gradlew adBwcCluster#fullRestartClusterTask -Dtests.security.manager=false` launches a cluster with three nodes of bwc version of OpenSearch with anomaly-detection and job-scheduler and tests backwards compatibility by performing a full restart on the cluster upgrading all the nodes with the current version of OpenSearch with anomaly-detection and job-scheduler.
+12. `./gradlew bwcTestSuite -Dtests.security.manager=false` runs all the above bwc tests combined.
 
 When launching a cluster using one of the above commands logs are placed in `/build/cluster/run node0/opensearch-<version>/logs`. Though the logs are teed to the console, in practices it's best to check the actual log file.
 
diff --git a/plugin/build.gradle b/plugin/build.gradle
index 956a5276dd..828b4b2c2d 100644
--- a/plugin/build.gradle
+++ b/plugin/build.gradle
@@ -138,6 +138,13 @@ integTest {
         }
     }
 
+    // BWC test can only run within the BWC gradle task bwcsuite or its dependent tasks.
+    if (System.getProperty('tests.rest.bwcsuite') == null) {
+        filter {
+            excludeTestsMatching "org.opensearch.ml.bwc.*IT"
+        }
+    }
+
     // The 'doFirst' delays till execution time.
     doFirst {
         // Tell the test JVM if the cluster JVM is running under a debugger so that tests can
@@ -371,3 +378,166 @@ tasks.withType(licenseHeaders.class) {
 checkstyle {
     toolVersion = '8.29'
 }
+
+String bwcVersion = "2.4.0.0"
+String bwcShortVersion = bwcVersion[0..4]
+String baseName = "mlCommonsBwcCluster"
+String bwcMlPlugin = "opensearch-ml-" + bwcVersion + ".zip"
+String bwcFilePath = "src/test/resources/org/opensearch/ml/bwc/"
+String bwcRemoteFile = "https://ci.opensearch.org/ci/dbc/distribution-build-opensearch/" + bwcShortVersion + "/latest/linux/x64/tar/builds/opensearch/plugins/" + bwcMlPlugin
+String project_no_snapshot = project.version.replace("-SNAPSHOT","")
+String opensearch_no_snapshot = opensearch_version.replace("-SNAPSHOT","")
+String opensearchMlPlugin = "opensearch-ml-" + project_no_snapshot + ".zip"
+String opensearchMlRemoteFile = 'https://ci.opensearch.org/ci/dbc/distribution-build-opensearch/' + opensearch_no_snapshot +
+        '/latest/linux/x64/tar/builds/opensearch/plugins/' + opensearchMlPlugin
+
+2.times {i ->
+    testClusters {
+        "${baseName}$i" {
+            testDistribution = "ARCHIVE"
+            versions = [bwcShortVersion, opensearch_version]
+            numberOfNodes = 3
+            plugin(provider(new Callable<RegularFile>() {
+                @Override
+                RegularFile call() throws Exception {
+                    return new RegularFile() {
+                        @Override
+                        File getAsFile() {
+                            File dir = new File('./plugin/' + bwcFilePath + bwcVersion)
+                            if (!dir.exists()) {
+                                dir.mkdirs()
+                            }
+                            File f = new File(dir, bwcMlPlugin)
+                            if (!f.exists()) {
+                                new URL(bwcRemoteFile).withInputStream{ ins -> f.withOutputStream{ it << ins }}
+                            }
+                            return fileTree(bwcFilePath + bwcVersion).getSingleFile()
+                        }
+                    }
+                }
+            }))
+            setting 'path.repo', "${buildDir}/cluster/shared/repo/${baseName}"
+            setting 'http.content_type.required', 'true'
+        }
+    }
+}
+
+List<Provider<RegularFile>> plugins = [
+        provider(new Callable<RegularFile>() {
+            @Override
+            RegularFile call() throws Exception {
+                return new RegularFile() {
+                    @Override
+                    File getAsFile() {
+                        if (new File('./plugin/' + bwcFilePath + project.version).exists()) {
+                            project.delete(files('./plugin/' + bwcFilePath + project.version))
+                        }
+                        project.mkdir bwcFilePath + project.version
+                        ant.get(src: opensearchMlRemoteFile,
+                                dest: bwcFilePath + project.version,
+                                httpusecaches: false)
+                        return fileTree(bwcFilePath + project.version).getSingleFile()
+                    }
+                }
+            }
+        })
+]
+
+// Creates 2 test clusters with 3 nodes of the old version.
+2.times { i ->
+    task "${baseName}#oldVersionClusterTask$i"(type: StandaloneRestIntegTestTask) {
+        useCluster testClusters."${baseName}$i"
+        filter {
+            includeTestsMatching "org.opensearch.ml.bwc.*IT"
+        }
+        systemProperty 'tests.rest.bwcsuite', 'old_cluster'
+        systemProperty 'tests.rest.bwcsuite_round', 'old'
+        systemProperty 'tests.plugin_bwc_version', bwcVersion
+        nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}$i".allHttpSocketURI.join(",")}")
+        nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}$i".getName()}")
+    }
+}
+
+// Upgrade one node of the old cluster to new OpenSearch version with upgraded plugin version
+// This results in a mixed cluster with 2 nodes on the old version and 1 upgraded node.
+// This is also used as a one third upgraded cluster for a rolling upgrade.
+task "${baseName}#mixedClusterTask"(type: StandaloneRestIntegTestTask) {
+    useCluster testClusters."${baseName}0"
+    dependsOn "${baseName}#oldVersionClusterTask0"
+    doFirst {
+        testClusters."${baseName}0".upgradeNodeAndPluginToNextVersion(plugins)
+    }
+    filter {
+        includeTestsMatching "org.opensearch.ml.bwc.*IT"
+    }
+    systemProperty 'tests.rest.bwcsuite', 'mixed_cluster'
+    systemProperty 'tests.rest.bwcsuite_round', 'first'
+    systemProperty 'tests.plugin_bwc_version', bwcVersion
+    nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}0".allHttpSocketURI.join(",")}")
+    nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}0".getName()}")
+}
+
+// Upgrades the second node to new OpenSearch version with upgraded plugin version after the first node is upgraded.
+// This results in a mixed cluster with 1 node on the old version and 2 upgraded nodes.
+// This is used for rolling upgrade.
+task "${baseName}#twoThirdsUpgradedClusterTask"(type: StandaloneRestIntegTestTask) {
+    dependsOn "${baseName}#mixedClusterTask"
+    useCluster testClusters."${baseName}0"
+    doFirst {
+        testClusters."${baseName}0".upgradeNodeAndPluginToNextVersion(plugins)
+    }
+    filter {
+        includeTestsMatching "org.opensearch.ml.bwc.*IT"
+    }
+    systemProperty 'tests.rest.bwcsuite', 'mixed_cluster'
+    systemProperty 'tests.rest.bwcsuite_round', 'second'
+    systemProperty 'tests.plugin_bwc_version', bwcVersion
+    nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}0".allHttpSocketURI.join(",")}")
+    nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}0".getName()}")
+}
+
+// Upgrade the third node to new OpenSearch version with upgraded plugin version after the second node is upgraded.
+// This results in a fully upgraded cluster.
+// This is used for rolling upgrade.
+task "${baseName}#rollingUpgradeClusterTask"(type: StandaloneRestIntegTestTask) {
+    dependsOn "${baseName}#twoThirdsUpgradedClusterTask"
+    useCluster testClusters."${baseName}0"
+    doFirst {
+        testClusters."${baseName}0".upgradeNodeAndPluginToNextVersion(plugins)
+    }
+    filter {
+        includeTestsMatching "org.opensearch.ml.bwc.*IT"
+    }
+    mustRunAfter "${baseName}#mixedClusterTask"
+    systemProperty 'tests.rest.bwcsuite', 'mixed_cluster'
+    systemProperty 'tests.rest.bwcsuite_round', 'third'
+    systemProperty 'tests.plugin_bwc_version', bwcVersion
+    nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}0".allHttpSocketURI.join(",")}")
+    nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}0".getName()}")
+}
+
+// Upgrades all the nodes of the old cluster to new OpenSearch version with upgraded plugin version
+// at the same time resulting in a fully upgraded cluster.
+task "${baseName}#fullRestartClusterTask"(type: StandaloneRestIntegTestTask) {
+    dependsOn "${baseName}#oldVersionClusterTask1"
+    useCluster testClusters."${baseName}1"
+    doFirst {
+        testClusters."${baseName}1".upgradeAllNodesAndPluginsToNextVersion(plugins)
+    }
+    filter {
+        includeTestsMatching "org.opensearch.ml.bwc.*IT"
+    }
+    systemProperty 'tests.rest.bwcsuite', 'upgraded_cluster'
+    systemProperty 'tests.plugin_bwc_version', bwcVersion
+    nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}1".allHttpSocketURI.join(",")}")
+    nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}1".getName()}")
+}
+
+// A bwc test suite which runs all the bwc tasks combined
+task bwcTestSuite(type: StandaloneRestIntegTestTask) {
+    exclude '**/*Test*'
+    exclude '**/*IT*'
+    dependsOn tasks.named("${baseName}#mixedClusterTask")
+    dependsOn tasks.named("${baseName}#rollingUpgradeClusterTask")
+    dependsOn tasks.named("${baseName}#fullRestartClusterTask")
+}
diff --git a/plugin/src/test/java/org/opensearch/ml/bwc/MLCommonsBackwardsCompatibilityIT.java b/plugin/src/test/java/org/opensearch/ml/bwc/MLCommonsBackwardsCompatibilityIT.java
new file mode 100644
index 0000000000..1bab83d828
--- /dev/null
+++ b/plugin/src/test/java/org/opensearch/ml/bwc/MLCommonsBackwardsCompatibilityIT.java
@@ -0,0 +1,227 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.ml.bwc;
+
+import static org.junit.Assert.*;
+import static org.opensearch.ml.common.input.parameter.clustering.KMeansParams.DistanceType.COSINE;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import org.apache.http.HttpEntity;
+import org.junit.Assume;
+import org.junit.Before;
+import org.opensearch.client.Response;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.index.query.MatchAllQueryBuilder;
+import org.opensearch.ml.common.FunctionName;
+import org.opensearch.ml.common.MLTaskState;
+import org.opensearch.ml.common.input.parameter.clustering.KMeansParams;
+import org.opensearch.ml.rest.MLCommonsRestTestCase;
+import org.opensearch.ml.utils.TestData;
+import org.opensearch.ml.utils.TestHelper;
+import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.test.rest.OpenSearchRestTestCase;
+
+public class MLCommonsBackwardsCompatibilityIT extends MLCommonsRestTestCase {
+
+    private final ClusterType CLUSTER_TYPE = ClusterType.parse(System.getProperty("tests.rest.bwcsuite"));
+    private final String CLUSTER_NAME = System.getProperty("tests.clustername");
+    private String MIXED_CLUSTER_TEST_ROUND = System.getProperty("tests.rest.bwcsuite_round");
+    private final String irisIndex = "iris_data_backwards_compatibility_it";
+    private SearchSourceBuilder searchSourceBuilder;
+    private KMeansParams kMeansParams;
+
+    @Before
+    public void setup() throws Exception {
+        Assume
+            .assumeTrue(
+                "Test cannot be run outside the BWC gradle task 'bwcTestSuite' or its dependent tasks",
+                System.getProperty("tests.rest.bwcsuite") != null
+            );
+        searchSourceBuilder = new SearchSourceBuilder();
+        searchSourceBuilder.query(new MatchAllQueryBuilder());
+        searchSourceBuilder.size(1000);
+        searchSourceBuilder.fetchSource(new String[] { "petal_length_in_cm", "petal_width_in_cm" }, null);
+
+        kMeansParams = KMeansParams.builder().centroids(3).iterations(10).distanceType(COSINE).build();
+    }
+
+    @Override
+    protected final boolean preserveIndicesUponCompletion() {
+        return true;
+    }
+
+    @Override
+    protected final boolean preserveReposUponCompletion() {
+        return true;
+    }
+
+    @Override
+    protected boolean preserveTemplatesUponCompletion() {
+        return true;
+    }
+
+    @Override
+    protected final Settings restClientSettings() {
+        return Settings
+            .builder()
+            .put(super.restClientSettings())
+            // increase the timeout here to 90 seconds to handle long waits for a green
+            // cluster health. the waits for green need to be longer than a minute to
+            // account for delayed shards
+            .put(OpenSearchRestTestCase.CLIENT_SOCKET_TIMEOUT, "90s")
+            .build();
+    }
+
+    private enum ClusterType {
+        OLD,
+        MIXED,
+        UPGRADED;
+
+        public static ClusterType parse(String value) {
+            switch (value) {
+                case "old_cluster":
+                    return OLD;
+                case "mixed_cluster":
+                    return MIXED;
+                case "upgraded_cluster":
+                    return UPGRADED;
+                default:
+                    throw new AssertionError("unknown cluster type: " + value);
+            }
+        }
+    }
+
+    private String getUri() {
+        switch (CLUSTER_TYPE) {
+            case OLD:
+                return "_nodes/" + CLUSTER_NAME + "-0/plugins";
+            case MIXED:
+                String round = System.getProperty("tests.rest.bwcsuite_round");
+                if (round.equals("second")) {
+                    return "_nodes/" + CLUSTER_NAME + "-1/plugins";
+                } else if (round.equals("third")) {
+                    return "_nodes/" + CLUSTER_NAME + "-2/plugins";
+                } else {
+                    return "_nodes/" + CLUSTER_NAME + "-0/plugins";
+                }
+            case UPGRADED:
+                return "_nodes/plugins";
+            default:
+                throw new AssertionError("unknown cluster type: " + CLUSTER_TYPE);
+        }
+    }
+
+    private int getMixedClusterTestRound() {
+        int mixedClusterTestRound = 0;
+        switch (MIXED_CLUSTER_TEST_ROUND) {
+            case "first":
+                mixedClusterTestRound = 1;
+                break;
+            case "second":
+                mixedClusterTestRound = 2;
+                break;
+            case "third":
+                mixedClusterTestRound = 3;
+                break;
+            default:
+                break;
+        }
+        return mixedClusterTestRound;
+    }
+
+    public void testBackwardsCompatibility() throws Exception {
+        String uri = getUri();
+        Map<String, Map<String, Object>> responseMap = (Map<String, Map<String, Object>>) getAsMap(uri).get("nodes");
+        for (Map<String, Object> response : responseMap.values()) {
+            List<Map<String, Object>> plugins = (List<Map<String, Object>>) response.get("plugins");
+            Set<Object> pluginNames = plugins.stream().map(map -> map.get("name")).collect(Collectors.toSet());
+            String opensearchVersion = plugins
+                .stream()
+                .map(map -> map.get("opensearch_version"))
+                .collect(Collectors.toSet())
+                .iterator()
+                .next()
+                .toString();
+            switch (CLUSTER_TYPE) {
+                case OLD:
+                    assertTrue(pluginNames.contains("opensearch-ml"));
+                    assertEquals("2.4.0", opensearchVersion);
+                    ingestIrisData(irisIndex);
+                    // train model
+                    train(client(), FunctionName.KMEANS, irisIndex, kMeansParams, searchSourceBuilder, trainResult -> {
+                        String modelId = (String) trainResult.get("model_id");
+                        assertNotNull(modelId);
+                        String status = (String) trainResult.get("status");
+                        assertEquals(MLTaskState.COMPLETED.name(), status);
+                    }, false);
+                case MIXED:
+                    assertTrue(pluginNames.contains("opensearch-ml"));
+                    // then predict with old model
+                    if (opensearchVersion.equals("2.4.0")) {
+                        String modelId = getModelIdWithFunctionName(FunctionName.KMEANS);
+                        predict(client(), FunctionName.KMEANS, modelId, irisIndex, kMeansParams, searchSourceBuilder, predictResult -> {
+                            String predictStatus = (String) predictResult.get("status");
+                            assertEquals(MLTaskState.COMPLETED.name(), predictStatus);
+                            Map<String, Object> predictionResult = (Map<String, Object>) predictResult.get("prediction_result");
+                            ArrayList rows = (ArrayList) predictionResult.get("rows");
+                            assertTrue(rows.size() > 1);
+                        });
+                    } else if (opensearchVersion.equals("2.5.0")) {
+                        // train predict with old data
+                        ingestIrisData(irisIndex);
+                        trainAndPredict(client(), FunctionName.KMEANS, irisIndex, kMeansParams, searchSourceBuilder, predictionResult -> {
+                            ArrayList rows = (ArrayList) predictionResult.get("rows");
+                            assertTrue(rows.size() > 0);
+                        });
+                    } else {
+                        throw new AssertionError("Cannot get the correct version for opensearch ml-commons plugin for the bwc test.");
+                    }
+                    break;
+                case UPGRADED:
+                    assertTrue(pluginNames.contains("opensearch-ml"));
+                    assertEquals("2.5.0", opensearchVersion);
+                    ingestIrisData(irisIndex);
+                    trainAndPredict(client(), FunctionName.KMEANS, irisIndex, kMeansParams, searchSourceBuilder, predictionResult -> {
+                        ArrayList rows = (ArrayList) predictionResult.get("rows");
+                        assertTrue(rows.size() > 0);
+                    });
+                    break;
+            }
+            break;
+        }
+    }
+
+    private String getModelIdWithFunctionName(FunctionName functionName) throws IOException {
+        String modelQuery = "{\"query\": {"
+            + "\"term\": {"
+            + "\"algorithm\":{\"value\": "
+            + "\""
+            + functionName.name().toUpperCase(Locale.ROOT)
+            + "\""
+            + "}"
+            + "}"
+            + "}"
+            + "}";
+        Response searchModelResponse = TestHelper.makeRequest(client(), "GET", "/_plugins/_ml/models/_search", null, modelQuery, null);
+        HttpEntity entity = searchModelResponse.getEntity();
+        String entityString = TestHelper.httpEntityToString(entity);
+        Map modelResponseMap = gson.fromJson(entityString, Map.class);
+        Map hitsModelsMap = (Map) modelResponseMap.get("hits");
+        List<Map<String, Object>> hitsModels = (List<Map<String, Object>>) hitsModelsMap.get("hits");
+        Set<Object> modelIdSet = hitsModels.stream().map(map -> map.get("_id")).collect(Collectors.toSet());
+        return modelIdSet.iterator().next().toString();
+    }
+
+    private void verifyMlResponse(String uri) throws Exception {
+        Response response = TestHelper.makeRequest(client(), "GET", uri, null, TestData.matchAllSearchQuery(), null);
+        HttpEntity entity = response.getEntity();
+        String entityString = TestHelper.httpEntityToString(entity);
+        assertNull(entityString);
+    }
+}