Skip to content

Commit 0f53dec

Browse files
mohitamgmohitamg
committed
Add force resume integration tests
Signed-off-by: Mohit Kumar <mohitamg@amazon.com>
1 parent a22e956 commit 0f53dec

2 files changed

Lines changed: 206 additions & 0 deletions

File tree

‎src/test/kotlin/org/opensearch/replication/ReplicationHelpers.kt‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -253,6 +253,17 @@ fun RestHighLevelClient.resumeReplication(index: String, requestOptions: Request
253253
waitForReplicationStart(index, TimeValue.timeValueSeconds(10))
254254
}
255255

256+
fun RestHighLevelClient.forceResumeReplication(index: String, requestOptions: RequestOptions = RequestOptions.DEFAULT) {
257+
val url = REST_REPLICATION_RESUME.replace("{index}", index, true)
258+
val lowLevelResumeRequest = Request("POST", url)
259+
lowLevelResumeRequest.setJsonEntity("""{"force_resume": true}""")
260+
lowLevelResumeRequest.setOptions(requestOptions)
261+
val lowLevelResumeResponse = lowLevelClient.performRequest(lowLevelResumeRequest)
262+
val response = getAckResponse(lowLevelResumeResponse)
263+
assertThat(response.isAcknowledged).withFailMessage("Replication could not be force resumed").isTrue()
264+
waitForReplicationStart(index, TimeValue.timeValueSeconds(60))
265+
}
266+
256267
fun RestHighLevelClient.updateReplication(index: String, settings: Settings, requestOptions: RequestOptions = RequestOptions.DEFAULT,
257268
clusterManagerTimeout: String? = null) {
258269
var url = REST_REPLICATION_UPDATE.replace("{index}", index, true)

‎src/test/kotlin/org/opensearch/replication/integ/rest/ForceResumeReplicationIT.kt‎

Lines changed: 195 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,195 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*
8+
* Modifications Copyright OpenSearch Contributors. See
9+
* GitHub history for details.
10+
*/
11+
12+
package org.opensearch.replication.integ.rest
13+
14+
import org.opensearch.replication.MultiClusterAnnotations
15+
import org.opensearch.replication.MultiClusterRestTestCase
16+
import org.opensearch.replication.StartReplicationRequest
17+
import org.opensearch.replication.forceResumeReplication
18+
import org.opensearch.replication.pauseReplication
19+
import org.opensearch.replication.replicationStatus
20+
import org.opensearch.replication.resumeReplication
21+
import org.opensearch.replication.startReplication
22+
import org.opensearch.replication.stopReplication
23+
import org.opensearch.replication.`validate paused status response`
24+
import org.opensearch.replication.`validate status syncing response`
25+
import org.assertj.core.api.Assertions.assertThat
26+
import org.assertj.core.api.Assertions.assertThatThrownBy
27+
import org.opensearch.action.DocWriteResponse
28+
import org.opensearch.action.admin.indices.delete.DeleteIndexRequest
29+
import org.opensearch.action.index.IndexRequest
30+
import org.opensearch.client.RequestOptions
31+
import org.opensearch.client.ResponseException
32+
import org.opensearch.client.indices.CreateIndexRequest
33+
import org.opensearch.client.indices.GetIndexRequest
34+
import org.junit.Assert
35+
import java.util.concurrent.TimeUnit
36+
37+
@MultiClusterAnnotations.ClusterConfigurations(
38+
MultiClusterAnnotations.ClusterConfiguration(clusterName = LEADER),
39+
MultiClusterAnnotations.ClusterConfiguration(clusterName = FOLLOWER)
40+
)
41+
class ForceResumeReplicationIT : MultiClusterRestTestCase() {
42+
private val leaderIndexName = "leader_index"
43+
private val followerIndexName = "force_resumed_index"
44+
45+
fun `test force resume after retention lease expires`() {
46+
val followerClient = getClientForCluster(FOLLOWER)
47+
val leaderClient = getClientForCluster(LEADER)
48+
createConnectionBetweenClusters(FOLLOWER, LEADER)
49+
50+
// Create leader index and start replication
51+
val createIndexResponse = leaderClient.indices().create(
52+
CreateIndexRequest(leaderIndexName), RequestOptions.DEFAULT
53+
)
54+
assertThat(createIndexResponse.isAcknowledged).isTrue()
55+
56+
followerClient.startReplication(
57+
StartReplicationRequest("source", leaderIndexName, followerIndexName),
58+
waitForRestore = true
59+
)
60+
61+
// Index some data on leader to advance the global checkpoint
62+
val sourceMap: MutableMap<String, String> = HashMap()
63+
sourceMap["field1"] = "value1"
64+
val indexResponse = leaderClient.index(
65+
IndexRequest(leaderIndexName).id("1").source(sourceMap), RequestOptions.DEFAULT
66+
)
67+
assertThat(indexResponse.result).isIn(DocWriteResponse.Result.CREATED, DocWriteResponse.Result.UPDATED)
68+
69+
// Wait for replication to sync
70+
assertBusy({
71+
val statusResp = followerClient.replicationStatus(followerIndexName)
72+
`validate status syncing response`(statusResp)
73+
}, 30, TimeUnit.SECONDS)
74+
75+
// Pause replication
76+
followerClient.pauseReplication(followerIndexName)
77+
val statusResp = followerClient.replicationStatus(followerIndexName)
78+
`validate paused status response`(statusResp)
79+
80+
// Simulate retention lease expiry by deleting and recreating the leader index.
81+
// This causes the retention leases to be lost.
82+
val deleteResponse = leaderClient.indices().delete(
83+
DeleteIndexRequest(leaderIndexName), RequestOptions.DEFAULT
84+
)
85+
assertThat(deleteResponse.isAcknowledged).isTrue()
86+
87+
val recreateResponse = leaderClient.indices().create(
88+
CreateIndexRequest(leaderIndexName), RequestOptions.DEFAULT
89+
)
90+
assertThat(recreateResponse.isAcknowledged).isTrue()
91+
92+
// Normal resume should fail because retention leases are gone
93+
assertThatThrownBy {
94+
followerClient.resumeReplication(followerIndexName)
95+
}.isInstanceOf(ResponseException::class.java)
96+
.hasMessageContaining("Retention lease doesn't exist")
97+
98+
// Force resume should succeed — triggers snapshot bootstrap
99+
followerClient.forceResumeReplication(followerIndexName)
100+
101+
// Verify replication is back in syncing state
102+
assertBusy({
103+
val syncStatus = followerClient.replicationStatus(followerIndexName)
104+
`validate status syncing response`(syncStatus)
105+
}, 60, TimeUnit.SECONDS)
106+
107+
// Verify the follower index exists and is functional
108+
val indexExists = followerClient.indices().exists(
109+
GetIndexRequest(followerIndexName), RequestOptions.DEFAULT
110+
)
111+
assertThat(indexExists).isTrue()
112+
113+
// Index more data on leader and verify it replicates
114+
sourceMap["field2"] = "value2"
115+
leaderClient.index(
116+
IndexRequest(leaderIndexName).id("2").source(sourceMap), RequestOptions.DEFAULT
117+
)
118+
119+
assertBusy({
120+
val count = followerClient.count(
121+
org.opensearch.client.core.CountRequest(followerIndexName), RequestOptions.DEFAULT
122+
).count
123+
assertThat(count).isGreaterThanOrEqualTo(1L)
124+
}, 60, TimeUnit.SECONDS)
125+
126+
// Cleanup
127+
followerClient.stopReplication(followerIndexName)
128+
}
129+
130+
fun `test force resume when retention leases still exist proceeds normally`() {
131+
val followerClient = getClientForCluster(FOLLOWER)
132+
val leaderClient = getClientForCluster(LEADER)
133+
createConnectionBetweenClusters(FOLLOWER, LEADER)
134+
135+
// Create leader index and start replication
136+
val createIndexResponse = leaderClient.indices().create(
137+
CreateIndexRequest(leaderIndexName), RequestOptions.DEFAULT
138+
)
139+
assertThat(createIndexResponse.isAcknowledged).isTrue()
140+
141+
followerClient.startReplication(
142+
StartReplicationRequest("source", leaderIndexName, followerIndexName),
143+
waitForRestore = true
144+
)
145+
146+
// Wait for syncing state
147+
assertBusy({
148+
val statusResp = followerClient.replicationStatus(followerIndexName)
149+
`validate status syncing response`(statusResp)
150+
}, 30, TimeUnit.SECONDS)
151+
152+
// Pause replication (retention leases are still valid)
153+
followerClient.pauseReplication(followerIndexName)
154+
155+
// Force resume with valid leases should succeed (falls through to normal resume)
156+
followerClient.forceResumeReplication(followerIndexName)
157+
158+
// Verify replication is back in syncing state
159+
assertBusy({
160+
val syncStatus = followerClient.replicationStatus(followerIndexName)
161+
`validate status syncing response`(syncStatus)
162+
}, 30, TimeUnit.SECONDS)
163+
164+
// Cleanup
165+
followerClient.stopReplication(followerIndexName)
166+
}
167+
168+
fun `test force resume error message suggests force_resume option`() {
169+
val followerClient = getClientForCluster(FOLLOWER)
170+
val leaderClient = getClientForCluster(LEADER)
171+
createConnectionBetweenClusters(FOLLOWER, LEADER)
172+
173+
// Create leader index and start replication
174+
val createIndexResponse = leaderClient.indices().create(
175+
CreateIndexRequest(leaderIndexName), RequestOptions.DEFAULT
176+
)
177+
assertThat(createIndexResponse.isAcknowledged).isTrue()
178+
179+
followerClient.startReplication(
180+
StartReplicationRequest("source", leaderIndexName, followerIndexName),
181+
waitForRestore = true
182+
)
183+
184+
// Pause and break retention leases
185+
followerClient.pauseReplication(followerIndexName)
186+
leaderClient.indices().delete(DeleteIndexRequest(leaderIndexName), RequestOptions.DEFAULT)
187+
leaderClient.indices().create(CreateIndexRequest(leaderIndexName), RequestOptions.DEFAULT)
188+
189+
// Normal resume should fail with a helpful error message
190+
assertThatThrownBy {
191+
followerClient.resumeReplication(followerIndexName)
192+
}.isInstanceOf(ResponseException::class.java)
193+
.hasMessageContaining("force_resume=true")
194+
}
195+
}

0 commit comments

Comments
 (0)