Skip to content

Commit

Permalink
Allow empty translog for ITs testing no ingestion post refresh/commit (…
Browse files Browse the repository at this point in the history 
…opensearch-project#11946)

Signed-off-by: bansvaru <[email protected]>
Signed-off-by: Shivansh Arora <[email protected]>
  • Loading branch information
@linuxpi @shiv0408
linuxpi authored and shiv0408 committed Apr 25, 2024
1 parent 641dd44 commit 40716ea
Show file tree
Hide file tree
Showing 2 changed files with 23 additions and 13 deletions.
9 changes: 9 additions & 0 deletions ...src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreBaseIntegTestCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,10 @@ public class RemoteStoreBaseIntegTestCase extends OpenSearchIntegTestCase {
);

protected Map<String, Long> indexData(int numberOfIterations, boolean invokeFlush, String index) {
return indexData(numberOfIterations, invokeFlush, false, index);
}

protected Map<String, Long> indexData(int numberOfIterations, boolean invokeFlush, boolean emptyTranslog, String index) {
long totalOperations = 0;
long refreshedOrFlushedOperations = 0;
long maxSeqNo = -1;
Expand All @@ -96,6 +100,11 @@ protected Map<String, Long> indexData(int numberOfIterations, boolean invokeFlus
} else {
refresh(index);
}

// skip indexing if last iteration as we dont want to have any data in remote translog
if (emptyTranslog && i == numberOfIterations - 1) {
continue;
}
maxSeqNoRefreshedOrFlushed = maxSeqNo;
indexingStats.put(MAX_SEQ_NO_REFRESHED_OR_FLUSHED + "-shard-" + shardId, maxSeqNoRefreshedOrFlushed);
refreshedOrFlushedOperations = totalOperations;
Expand Down
27 changes: 14 additions & 13 deletions server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreRestoreIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,15 +44,15 @@ public class RemoteStoreRestoreIT extends BaseRemoteStoreRestoreIT {
* @throws IOException IO Exception.
*/
public void testRemoteTranslogRestoreWithNoDataPostCommit() throws Exception {
testRestoreFlow(1, true, randomIntBetween(1, 5));
testRestoreFlow(1, true, true, randomIntBetween(1, 5));
}

/**
* Simulates all data restored using Remote Translog Store.
* @throws IOException IO Exception.
*/
public void testRemoteTranslogRestoreWithNoDataPostRefresh() throws Exception {
testRestoreFlow(1, false, randomIntBetween(1, 5));
testRestoreFlow(1, false, true, randomIntBetween(1, 5));
}

/**
Expand All @@ -61,7 +61,7 @@ public void testRemoteTranslogRestoreWithNoDataPostRefresh() throws Exception {
* @throws IOException IO Exception.
*/
public void testRemoteTranslogRestoreWithRefreshedData() throws Exception {
testRestoreFlow(randomIntBetween(2, 5), false, randomIntBetween(1, 5));
testRestoreFlow(randomIntBetween(2, 5), false, false, randomIntBetween(1, 5));
}

/**
Expand All @@ -70,23 +70,23 @@ public void testRemoteTranslogRestoreWithRefreshedData() throws Exception {
* @throws IOException IO Exception.
*/
public void testRemoteTranslogRestoreWithCommittedData() throws Exception {
testRestoreFlow(randomIntBetween(2, 5), true, randomIntBetween(1, 5));
testRestoreFlow(randomIntBetween(2, 5), true, false, randomIntBetween(1, 5));
}

/**
* Simulates all data restored using Remote Translog Store.
* @throws IOException IO Exception.
*/
public void testRTSRestoreWithNoDataPostCommitPrimaryReplicaDown() throws Exception {
testRestoreFlowBothPrimaryReplicasDown(1, true, randomIntBetween(1, 5));
testRestoreFlowBothPrimaryReplicasDown(1, true, true, randomIntBetween(1, 5));
}

/**
* Simulates all data restored using Remote Translog Store.
* @throws IOException IO Exception.
*/
public void testRTSRestoreWithNoDataPostRefreshPrimaryReplicaDown() throws Exception {
testRestoreFlowBothPrimaryReplicasDown(1, false, randomIntBetween(1, 5));
testRestoreFlowBothPrimaryReplicasDown(1, false, true, randomIntBetween(1, 5));
}

/**
Expand All @@ -95,7 +95,7 @@ public void testRTSRestoreWithNoDataPostRefreshPrimaryReplicaDown() throws Excep
* @throws IOException IO Exception.
*/
public void testRTSRestoreWithRefreshedDataPrimaryReplicaDown() throws Exception {
testRestoreFlowBothPrimaryReplicasDown(randomIntBetween(2, 5), false, randomIntBetween(1, 5));
testRestoreFlowBothPrimaryReplicasDown(randomIntBetween(2, 5), false, false, randomIntBetween(1, 5));
}

/**
Expand All @@ -104,7 +104,7 @@ public void testRTSRestoreWithRefreshedDataPrimaryReplicaDown() throws Exception
* @throws IOException IO Exception.
*/
public void testRTSRestoreWithCommittedDataPrimaryReplicaDown() throws Exception {
testRestoreFlowBothPrimaryReplicasDown(randomIntBetween(2, 5), true, randomIntBetween(1, 5));
testRestoreFlowBothPrimaryReplicasDown(randomIntBetween(2, 5), true, false, randomIntBetween(1, 5));
}

private void restoreAndVerify(int shardCount, int replicaCount, Map<String, Long> indexStats) throws Exception {
Expand All @@ -122,9 +122,9 @@ private void restoreAndVerify(int shardCount, int replicaCount, Map<String, Long
* @param invokeFlush If true, a flush is invoked. Otherwise, a refresh is invoked.
* @throws IOException IO Exception.
*/
private void testRestoreFlow(int numberOfIterations, boolean invokeFlush, int shardCount) throws Exception {
private void testRestoreFlow(int numberOfIterations, boolean invokeFlush, boolean emptyTranslog, int shardCount) throws Exception {
prepareCluster(1, 3, INDEX_NAME, 0, shardCount);
Map<String, Long> indexStats = indexData(numberOfIterations, invokeFlush, INDEX_NAME);
Map<String, Long> indexStats = indexData(numberOfIterations, invokeFlush, emptyTranslog, INDEX_NAME);
assertEquals(shardCount, getNumShards(INDEX_NAME).totalNumShards);

assertHitCount(client().prepareSearch(INDEX_NAME).setSize(0).get(), indexStats.get(REFRESHED_OR_FLUSHED_OPERATIONS));
Expand All @@ -141,9 +141,10 @@ private void testRestoreFlow(int numberOfIterations, boolean invokeFlush, int sh
* @param invokeFlush If true, a flush is invoked. Otherwise, a refresh is invoked.
* @throws IOException IO Exception.
*/
private void testRestoreFlowBothPrimaryReplicasDown(int numberOfIterations, boolean invokeFlush, int shardCount) throws Exception {
private void testRestoreFlowBothPrimaryReplicasDown(int numberOfIterations, boolean invokeFlush, boolean emptyTranslog, int shardCount)
throws Exception {
prepareCluster(1, 2, INDEX_NAME, 1, shardCount);
Map<String, Long> indexStats = indexData(numberOfIterations, invokeFlush, INDEX_NAME);
Map<String, Long> indexStats = indexData(numberOfIterations, invokeFlush, emptyTranslog, INDEX_NAME);
assertEquals(shardCount * 2, getNumShards(INDEX_NAME).totalNumShards);

internalCluster().stopRandomNode(InternalTestCluster.nameFilter(replicaNodeName(INDEX_NAME)));
Expand Down Expand Up @@ -391,7 +392,7 @@ public void testRTSRestoreWithCommittedDataExcludeIndicesPatterns() throws Excep
* @throws IOException IO Exception.
*/
public void testRTSRestoreDataOnlyInTranslog() throws Exception {
testRestoreFlow(0, true, randomIntBetween(1, 5));
testRestoreFlow(0, true, false, randomIntBetween(1, 5));
}

public void testRateLimitedRemoteDownloads() throws Exception {
Expand Down

0 comments on commit 40716ea

Please sign in to comment.