Updating filters as well during Alias update (#491) (#526)

Testing : Integ Test, Local
Signed-off-by: Gaurav Bafna <gbbafna@amazon.com>

Author: gbbafna

src/main/kotlin/org/opensearch/replication/task/index/IndexReplicationTask.kt

@@ -518,13 +518,18 @@ open class IndexReplicationTask(id: Long, type: String, action: String, descript
                     for (alias in toAdd) {
                         log.info("Adding alias ${alias.alias} from $followerIndexName")
                         // Copying writeIndex from leader doesn't cause any issue as writes will be blocked anyways
-                        request.addAliasAction(AliasActions.add().index(followerIndexName)
-                                .alias(alias.alias)
-                                .indexRouting(alias.indexRouting)
-                                .searchRouting(alias.searchRouting)
-                                .writeIndex(alias.writeIndex())
-                                .isHidden(alias.isHidden)
-                        )
+                        var aliasAction = AliasActions.add().index(followerIndexName)
+                            .alias(alias.alias)
+                            .indexRouting(alias.indexRouting)
+                            .searchRouting(alias.searchRouting)
+                            .writeIndex(alias.writeIndex())
+                            .isHidden(alias.isHidden)
+
+                        if (alias.filteringRequired())  {
+                            aliasAction = aliasAction.filter(alias.filter.string())
+                        }
+
+                        request.addAliasAction(aliasAction)
                     }
 
                     var toRemove = followerAliases - leaderAliases

src/test/kotlin/org/opensearch/replication/integ/rest/StartReplicationIT.kt

@@ -330,7 +330,9 @@ class StartReplicationIT: MultiClusterRestTestCase() {
 
         createConnectionBetweenClusters(FOLLOWER, LEADER)
 
-        val createIndexResponse = leaderClient.indices().create(CreateIndexRequest(leaderIndexName).alias(Alias("leaderAlias")), RequestOptions.DEFAULT)
+        val createIndexResponse = leaderClient.indices().create(CreateIndexRequest(leaderIndexName)
+            .alias(Alias("leaderAlias").filter("{\"term\":{\"year\":2016}}").routing("1"))
+            , RequestOptions.DEFAULT)
         assertThat(createIndexResponse.isAcknowledged).isTrue()
         try {
             followerClient.startReplication(StartReplicationRequest("source", leaderIndexName, followerIndexName))
@@ -339,12 +341,15 @@ class StartReplicationIT: MultiClusterRestTestCase() {
                         .exists(GetIndexRequest(followerIndexName), RequestOptions.DEFAULT))
                         .isEqualTo(true)
             }
-            Assert.assertEquals(
-                    leaderClient.indices().getAlias(GetAliasesRequest().indices(leaderIndexName),
-                            RequestOptions.DEFAULT).aliases[leaderIndexName],
-                    followerClient.indices().getAlias(GetAliasesRequest().indices(followerIndexName),
-                            RequestOptions.DEFAULT).aliases[followerIndexName]
-            )
+            assertBusy({
+                Assert.assertEquals(
+                        leaderClient.indices().getAlias(GetAliasesRequest().indices(leaderIndexName),
+                                RequestOptions.DEFAULT).aliases[leaderIndexName],
+                        followerClient.indices().getAlias(GetAliasesRequest().indices(followerIndexName),
+                                RequestOptions.DEFAULT).aliases[followerIndexName]
+                )
+
+            }, 30L, TimeUnit.SECONDS)
         } finally {
             followerClient.stopReplication(followerIndexName)
         }
@@ -521,7 +526,7 @@ class StartReplicationIT: MultiClusterRestTestCase() {
             var indicesAliasesRequest = IndicesAliasesRequest()
             var aliasAction = IndicesAliasesRequest.AliasActions.add()
                     .index(leaderIndexName)
-                    .alias("alias1")
+                    .alias("alias1").filter("{\"term\":{\"year\":2016}}").routing("1")
             indicesAliasesRequest.addAliasAction(aliasAction)
             leaderClient.indices().updateAliases(indicesAliasesRequest, RequestOptions.DEFAULT)