diff --git a/.github/workflows/multi-node-test-workflow.yml b/.github/workflows/multi-node-test-workflow.yml
index 3511f7d9a..172800a2d 100644
--- a/.github/workflows/multi-node-test-workflow.yml
+++ b/.github/workflows/multi-node-test-workflow.yml
@@ -78,3 +78,8 @@ jobs:
           else
             echo "Security plugin is NOT available skipping this run as tests without security have already been run"
           fi
+
+      - name: Run Alerting Backwards Compatibility Tests
+        run: |
+          echo "Running backwards compatibility tests ..."
+          ./gradlew bwcTestSuite
diff --git a/DEVELOPER_GUIDE.md b/DEVELOPER_GUIDE.md
index 93f36331b..282d23e93 100644
--- a/DEVELOPER_GUIDE.md
+++ b/DEVELOPER_GUIDE.md
@@ -44,6 +44,10 @@ Currently we just put RCF jar in lib as dependency. Plan to publish to Maven and
 6. `./gradlew :alerting:integTest -Dtests.class="*MonitorRunnerIT"` runs a single integ test class
 7. `./gradlew :alerting:integTest -Dtests.method="test execute monitor with dryrun"` runs a single integ test method
  (remember to quote the test method name if it contains spaces).
+8. `./gradlew alertingBwcCluster#mixedClusterTask` launches a cluster with three nodes of bwc version of OpenSearch with alerting and tests backwards compatibility by upgrading one of the nodes with the current version of OpenSearch with alerting, creating a mixed cluster.
+9. `./gradlew alertingBwcCluster#rollingUpgradeClusterTask` launches a cluster with three nodes of bwc version of OpenSearch with alerting and tests backwards compatibility by performing rolling upgrade of each node with the current version of OpenSearch with alerting.
+10. `./gradlew alertingBwcCluster#fullRestartClusterTask` launches a cluster with three nodes of bwc version of OpenSearch with alerting and tests backwards compatibility by performing a full restart on the cluster upgrading all the nodes with the current version of OpenSearch with alerting.
+11. `./gradlew bwcTestSuite` runs all the above bwc tests combined.
 
 When launching a cluster using one of the above commands, logs are placed in `alerting/build/testclusters/integTest-0/logs/`. Though the logs are teed to the console, in practices it's best to check the actual log file.
 
diff --git a/alerting/build.gradle b/alerting/build.gradle
index f52f9cfa1..761304b1e 100644
--- a/alerting/build.gradle
+++ b/alerting/build.gradle
@@ -24,7 +24,9 @@
  *   permissions and limitations under the License.
  */
 
+import java.util.concurrent.Callable
 import org.opensearch.gradle.test.RestIntegTestTask
+import org.opensearch.gradle.testclusters.StandaloneRestIntegTestTask
 
 apply plugin: 'java'
 apply plugin: 'idea'
@@ -164,6 +166,151 @@ integTest {
     if (System.getProperty("test.debug") != null) {
         jvmArgs '-agentlib:jdwp=transport=dt_socket,server=n,suspend=y,address=8000'
     }
+
+    if (System.getProperty("tests.rest.bwcsuite") == null) {
+        filter {
+            excludeTestsMatching "org.opensearch.alerting.bwc.*IT"
+        }
+    }
+}
+
+String bwcVersion = "1.13.1.0"
+String baseName = "alertingBwcCluster"
+String bwcFilePath = "src/test/resources/bwc"
+
+2.times {i ->
+    testClusters {
+        "${baseName}$i" {
+            testDistribution = "ARCHIVE"
+            versions = ["7.10.2","1.2.0-SNAPSHOT"]
+            numberOfNodes = 3
+            plugin(provider(new Callable<RegularFile>(){
+                @Override
+                RegularFile call() throws Exception {
+                    return new RegularFile() {
+                        @Override
+                        File getAsFile() {
+                            return fileTree(bwcFilePath + "/alerting/" + bwcVersion).getSingleFile()
+                        }
+                    }
+                }
+            }))
+            setting 'path.repo', "${buildDir}/cluster/shared/repo/${baseName}"
+            setting 'http.content_type.required', 'true'
+        }
+    }
+}
+
+List<Provider<RegularFile>> plugins = []
+
+// Ensure the artifact for the current project version is available to be used for the bwc tests
+task prepareBwcTests {
+    dependsOn bundle
+    doLast {
+        plugins = [
+                project.getObjects().fileProperty().value(bundle.getArchiveFile())
+        ]
+    }
+}
+
+// Create two test clusters with 3 nodes of the old version
+2.times {i ->
+    task "${baseName}#oldVersionClusterTask$i"(type: StandaloneRestIntegTestTask) {
+        dependsOn 'prepareBwcTests'
+        useCluster testClusters."${baseName}$i"
+        filter {
+            includeTestsMatching "org.opensearch.alerting.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.alerting.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()}")
+}
+
+// Upgrade 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.alerting.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.alerting.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()}")
+}
+
+// Upgrade 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.alerting.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")
 }
 
 run {
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt b/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt
index 83a0efc11..b48db925e 100644
--- a/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt
@@ -85,9 +85,10 @@ import javax.management.remote.JMXServiceURL
 
 abstract class AlertingRestTestCase : ODFERestTestCase() {
 
-    private val isDebuggingTest = DisableOnDebug(null).isDebugging
-    private val isDebuggingRemoteCluster = System.getProperty("cluster.debug", "false")!!.toBoolean()
-    val numberOfNodes = System.getProperty("cluster.number_of_nodes", "1")!!.toInt()
+    protected val isDebuggingTest = DisableOnDebug(null).isDebugging
+    protected val isDebuggingRemoteCluster = System.getProperty("cluster.debug", "false")!!.toBoolean()
+    protected val numberOfNodes = System.getProperty("cluster.number_of_nodes", "1")!!.toInt()
+    protected val isMultiNode = numberOfNodes > 1
 
     override fun xContentRegistry(): NamedXContentRegistry {
         return NamedXContentRegistry(
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/ODFERestTestCase.kt b/alerting/src/test/kotlin/org/opensearch/alerting/ODFERestTestCase.kt
index 5c4081b35..700ba8cf2 100644
--- a/alerting/src/test/kotlin/org/opensearch/alerting/ODFERestTestCase.kt
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/ODFERestTestCase.kt
@@ -78,9 +78,13 @@ abstract class ODFERestTestCase : OpenSearchRestTestCase() {
         return true
     }
 
+    open fun preserveODFEIndicesAfterTest(): Boolean = false
+
     @Throws(IOException::class)
     @After
     open fun wipeAllODFEIndices() {
+        if (preserveODFEIndicesAfterTest()) return
+
         val response = client().performRequest(Request("GET", "/_cat/indices?format=json&expand_wildcards=all"))
 
         val xContentType = XContentType.fromMediaTypeOrFormat(response.entity.contentType.value)
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/bwc/AlertingBackwardsCompatibilityIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/bwc/AlertingBackwardsCompatibilityIT.kt
new file mode 100644
index 000000000..3f8f431ac
--- /dev/null
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/bwc/AlertingBackwardsCompatibilityIT.kt
@@ -0,0 +1,220 @@
+/*
+ * 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.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+
+package org.opensearch.alerting.bwc
+
+import org.apache.http.entity.ContentType.APPLICATION_JSON
+import org.apache.http.entity.StringEntity
+import org.opensearch.alerting.ALERTING_BASE_URI
+import org.opensearch.alerting.AlertingRestTestCase
+import org.opensearch.alerting.LEGACY_OPENDISTRO_ALERTING_BASE_URI
+import org.opensearch.alerting.makeRequest
+import org.opensearch.alerting.model.Monitor
+import org.opensearch.common.settings.Settings
+import org.opensearch.common.xcontent.XContentType
+import org.opensearch.index.query.QueryBuilders
+import org.opensearch.rest.RestStatus
+import org.opensearch.search.builder.SearchSourceBuilder
+
+class AlertingBackwardsCompatibilityIT : AlertingRestTestCase() {
+
+    companion object {
+        private val CLUSTER_TYPE = ClusterType.parse(System.getProperty("tests.rest.bwcsuite"))
+        private val CLUSTER_NAME = System.getProperty("tests.clustername")
+    }
+
+    override fun preserveIndicesUponCompletion(): Boolean = true
+
+    override fun preserveReposUponCompletion(): Boolean = true
+
+    override fun preserveTemplatesUponCompletion(): Boolean = true
+
+    override fun preserveODFEIndicesAfterTest(): Boolean = true
+
+    override fun restClientSettings(): Settings {
+        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(CLIENT_SOCKET_TIMEOUT, "90s")
+            .build()
+    }
+
+    @Throws(Exception::class)
+    @Suppress("UNCHECKED_CAST")
+    fun `test backwards compatibility`() {
+        val uri = getPluginUri()
+        val responseMap = getAsMap(uri)["nodes"] as Map<String, Map<String, Any>>
+        for (response in responseMap.values) {
+            val plugins = response["plugins"] as List<Map<String, Any>>
+            val pluginNames = plugins.map { plugin -> plugin["name"] }.toSet()
+            when (CLUSTER_TYPE) {
+                ClusterType.OLD -> {
+                    assertTrue(pluginNames.contains("opendistro-alerting"))
+                    createBasicMonitor()
+                }
+                ClusterType.MIXED -> {
+                    assertTrue(pluginNames.contains("opensearch-alerting"))
+                    verifyMonitorExists(LEGACY_OPENDISTRO_ALERTING_BASE_URI)
+                    // Waiting a minute to ensure the Monitor ran again at least once before checking if the job is running
+                    // on time
+                    // TODO: Should probably change the next execution time of the Monitor manually instead since this inflates
+                    //  the test execution by a lot
+                    Thread.sleep(60000)
+                    // TODO: Need to move the base URI being used here into a constant and rename ALERTING_BASE_URI to
+                    //  MONITOR_BASE_URI
+                    verifyMonitorStats("/_opendistro/_alerting")
+                }
+                ClusterType.UPGRADED -> {
+                    assertTrue(pluginNames.contains("opensearch-alerting"))
+                    verifyMonitorExists(ALERTING_BASE_URI)
+                    Thread.sleep(60000)
+                    verifyMonitorStats("/_plugins/_alerting")
+                }
+            }
+            break
+        }
+    }
+
+    private enum class ClusterType {
+        OLD,
+        MIXED,
+        UPGRADED;
+
+        companion object {
+            fun parse(value: String): ClusterType {
+                return when (value) {
+                    "old_cluster" -> OLD
+                    "mixed_cluster" -> MIXED
+                    "upgraded_cluster" -> UPGRADED
+                    else -> throw AssertionError("Unknown cluster type: $value")
+                }
+            }
+        }
+    }
+
+    private fun getPluginUri(): String {
+        return when (CLUSTER_TYPE) {
+            ClusterType.OLD -> "_nodes/$CLUSTER_NAME-0/plugins"
+            ClusterType.MIXED -> {
+                when (System.getProperty("tests.rest.bwcsuite_round")) {
+                    "second" -> "_nodes/$CLUSTER_NAME-1/plugins"
+                    "third" -> "_nodes/$CLUSTER_NAME-2/plugins"
+                    else -> "_nodes/$CLUSTER_NAME-0/plugins"
+                }
+            }
+            ClusterType.UPGRADED -> "_nodes/plugins"
+        }
+    }
+
+    @Throws(Exception::class)
+    private fun createBasicMonitor() {
+        val indexName = "test_bwc_index"
+        val legacyMonitorString = """
+            {
+              "type": "monitor",
+              "name": "test_bwc_monitor",
+              "enabled": true,
+              "schedule": {
+                "period": {
+                  "interval": 1,
+                  "unit": "MINUTES"
+                }
+              },
+              "inputs": [
+                {
+                  "search": {
+                    "indices": [
+                      "$indexName"
+                    ],
+                    "query": {
+                      "size": 0,
+                      "query": {
+                        "match_all": {}
+                      }
+                    }
+                  }
+                }
+              ],
+              "triggers": [
+                {
+                  "name": "abc",
+                  "severity": "1",
+                  "condition": {
+                    "script": {
+                      "source": "ctx.results[0].hits.total.value > 100000",
+                      "lang": "painless"
+                    }
+                  },
+                  "actions": []
+                }
+              ]
+            }
+        """.trimIndent()
+        createIndex(indexName, Settings.EMPTY)
+
+        val createResponse = client().makeRequest(
+            method = "POST",
+            endpoint = "$LEGACY_OPENDISTRO_ALERTING_BASE_URI?refresh=true",
+            params = emptyMap(),
+            entity = StringEntity(legacyMonitorString, APPLICATION_JSON)
+        )
+
+        assertEquals("Create monitor failed", RestStatus.CREATED, createResponse.restStatus())
+        val responseBody = createResponse.asMap()
+        val createdId = responseBody["_id"] as String
+        val createdVersion = responseBody["_version"] as Int
+        assertNotEquals("Create monitor response is missing id", Monitor.NO_ID, createdId)
+        assertTrue("Create monitor reponse has incorrect version", createdVersion > 0)
+    }
+
+    @Throws(Exception::class)
+    @Suppress("UNCHECKED_CAST")
+    private fun verifyMonitorExists(uri: String) {
+        val search = SearchSourceBuilder().query(QueryBuilders.matchAllQuery()).toString()
+        val searchResponse = client().makeRequest(
+            "GET",
+            "$uri/_search",
+            emptyMap(),
+            StringEntity(search, APPLICATION_JSON)
+        )
+        assertEquals("Search monitor failed", RestStatus.OK, searchResponse.restStatus())
+        val xcp = createParser(XContentType.JSON.xContent(), searchResponse.entity.content)
+        val hits = xcp.map()["hits"]!! as Map<String, Map<String, Any>>
+        val numberDocsFound = hits["total"]?.get("value")
+        assertEquals("Unexpected number of Monitors returned", 1, numberDocsFound)
+    }
+
+    @Throws(Exception::class)
+    @Suppress("UNCHECKED_CAST")
+    /**
+     * Monitor stats will check if the Monitor scheduled job is running on time but does not necessarily mean that the
+     * Monitor execution itself did not fail.
+     */
+    private fun verifyMonitorStats(uri: String) {
+        val statsResponse = client().makeRequest(
+            "GET",
+            "$uri/stats",
+            emptyMap()
+        )
+        assertEquals("Monitor stats failed", RestStatus.OK, statsResponse.restStatus())
+        val xcp = createParser(XContentType.JSON.xContent(), statsResponse.entity.content)
+        val responseMap = xcp.map()
+        val nodesCount = responseMap["_nodes"]!! as Map<String, Any>
+        val totalNodes = nodesCount["total"]
+        val successfulNodes = nodesCount["successful"]
+        val nodesOnSchedule = responseMap["nodes_on_schedule"]!!
+        assertEquals("Incorrect number of total nodes", 3, totalNodes)
+        assertEquals("Some nodes in stats response failed", totalNodes, successfulNodes)
+        assertEquals("Not all nodes are on schedule", totalNodes, nodesOnSchedule)
+    }
+}
diff --git a/alerting/src/test/resources/bwc/alerting/1.13.1.0/opendistro-alerting-1.13.1.0.zip b/alerting/src/test/resources/bwc/alerting/1.13.1.0/opendistro-alerting-1.13.1.0.zip
new file mode 100644
index 000000000..37f0e4f83
Binary files /dev/null and b/alerting/src/test/resources/bwc/alerting/1.13.1.0/opendistro-alerting-1.13.1.0.zip differ
diff --git a/build-tools/repositories.gradle b/build-tools/repositories.gradle
index 3a3957d4d..67e5e47ed 100644
--- a/build-tools/repositories.gradle
+++ b/build-tools/repositories.gradle
@@ -29,5 +29,4 @@ repositories {
     maven { url "https://aws.oss.sonatype.org/content/repositories/snapshots" }
     mavenCentral()
     jcenter()
-    maven { url "https://aws.oss.sonatype.org/content/repositories/snapshots" }
 }
\ No newline at end of file
diff --git a/build.gradle b/build.gradle
index 49a17decc..ff859d31c 100644
--- a/build.gradle
+++ b/build.gradle
@@ -41,7 +41,6 @@ buildscript {
         mavenCentral()
         maven { url "https://plugins.gradle.org/m2/" }
         jcenter()
-        maven { url "https://aws.oss.sonatype.org/content/repositories/snapshots" }
     }
     dependencies {
         classpath "org.opensearch.gradle:build-tools:${opensearch_version}"