Skip to content

Commit

Permalink
Fix flaky RemoteFSTranslogTests testMetadataFileDeletion test opensea…
Browse files Browse the repository at this point in the history 
…rch-project#9605 (opensearch-project#11899)

Signed-off-by: Ashish Singh <[email protected]>
  • Loading branch information
@ashking94 @rayshrey
ashking94 authored and rayshrey committed Mar 18, 2024
1 parent 2a87a94 commit a81a5de
Showing 1 changed file with 11 additions and 2 deletions.
13 changes: 11 additions & 2 deletions server/src/test/java/org/opensearch/index/translog/RemoteFsTranslogTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,6 @@
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
Expand Down Expand Up @@ -715,6 +714,7 @@ public void testSimpleOperationsUpload() throws Exception {
translog.setMinSeqNoToKeep(0);
// This should not trim anything from local
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));
assertEquals(2, translog.readers.size());
assertBusy(() -> {
assertEquals(4, translog.allUploaded().size());
Expand All @@ -728,6 +728,7 @@ public void testSimpleOperationsUpload() throws Exception {
// This should not trim tlog-2.* files from remote as we not uploading any more translog to remote
translog.setMinSeqNoToKeep(1);
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));
assertEquals(1, translog.readers.size());
assertBusy(() -> {
assertEquals(4, translog.allUploaded().size());
Expand Down Expand Up @@ -766,6 +767,7 @@ public void testMetadataFileDeletion() throws Exception {
addToTranslogAndListAndUpload(translog, ops, new Translog.Index(String.valueOf(i), i, primaryTerm.get(), new byte[] { 1 }));
translog.setMinSeqNoToKeep(i);
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));
assertEquals(1, translog.readers.size());
}
assertBusy(() -> assertEquals(4, translog.allUploaded().size()));
Expand All @@ -776,13 +778,15 @@ public void testMetadataFileDeletion() throws Exception {
addToTranslogAndListAndUpload(translog, ops, new Translog.Index(String.valueOf(i), i, primaryTerm.get(), new byte[] { 1 }));
}
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));
assertEquals(1 + moreDocs, translog.readers.size());
assertBusy(() -> assertEquals(2 + 2L * moreDocs, translog.allUploaded().size()));
assertBusy(() -> assertEquals(1, blobStoreTransferService.listAll(getTranslogDirectory().add(METADATA_DIR)).size()));

int totalDocs = numDocs + moreDocs;
translog.setMinSeqNoToKeep(totalDocs - 1);
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));

addToTranslogAndListAndUpload(
translog,
Expand All @@ -791,6 +795,7 @@ public void testMetadataFileDeletion() throws Exception {
);
translog.setMinSeqNoToKeep(totalDocs);
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));
assertBusy(() -> assertEquals(1, blobStoreTransferService.listAll(getTranslogDirectory().add(METADATA_DIR)).size()));

// Change primary term and test the deletion of older primaries
Expand Down Expand Up @@ -841,6 +846,7 @@ public void testDrainSync() throws Exception {

translog.setMinSeqNoToKeep(0);
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));
assertEquals(1, translog.readers.size());

// Case 1 - During ongoing uploads, the available permits are 0.
Expand Down Expand Up @@ -869,6 +875,7 @@ public void testDrainSync() throws Exception {
// Case 3 - After drainSync, if trimUnreferencedReaders is attempted, we do not delete from remote store.
translog.setMinSeqNoToKeep(1);
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));
assertEquals(1, translog.readers.size());
assertEquals(6, translog.allUploaded().size());
assertEquals(mdFiles, blobStoreTransferService.listAll(getTranslogDirectory().add(METADATA_DIR)));
Expand All @@ -892,6 +899,7 @@ public void testDrainSync() throws Exception {

translog.setMinSeqNoToKeep(3);
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));
assertEquals(1, translog.readers.size());
assertBusy(() -> assertEquals(4, translog.allUploaded().size()));
assertBusy(() -> assertEquals(1, blobStoreTransferService.listAll(getTranslogDirectory().add(METADATA_DIR)).size()));
Expand Down Expand Up @@ -1048,7 +1056,7 @@ public void testConcurrentWriteViewsAndSnapshot() throws Throwable {
final int threadId = i;
writers[i] = new Thread(new AbstractRunnable() {
@Override
public void doRun() throws BrokenBarrierException, InterruptedException, IOException {
public void doRun() throws Exception {
barrier.await();
int counter = 0;
while (run.get() && idGenerator.get() < maxOps) {
Expand Down Expand Up @@ -1090,6 +1098,7 @@ public void doRun() throws BrokenBarrierException, InterruptedException, IOExcep
// deletionPolicy.setLocalCheckpointOfSafeCommit(localCheckpoint);
translog.setMinSeqNoToKeep(localCheckpoint + 1);
translog.trimUnreferencedReaders();
assertBusy(() -> assertTrue(translog.isRemoteGenerationDeletionPermitsAvailable()));
}
}
if (id % 7 == 0) {
Expand Down

0 comments on commit a81a5de

Please sign in to comment.