PR comments

opensearch-project · Jul 7, 2023 · e512bb1 · e512bb1
1 parent e242ead
commit e512bb1
Show file tree

Hide file tree

Showing 4 changed files with 43 additions and 11 deletions.
diff --git a/server/src/main/java/org/opensearch/index/translog/transfer/BlobStoreTransferService.java b/server/src/main/java/org/opensearch/index/translog/transfer/BlobStoreTransferService.java
@@ -133,7 +133,7 @@ public void listFoldersAsync(String threadpoolName, Iterable<String> path, Actio
         });
     }
 
-    public void listAllInSortedOrder(Iterable<String> path, int limit, ActionListener<List<BlobMetadata>> listener) throws IOException {
+    public void listAllInSortedOrder(Iterable<String> path, int limit, ActionListener<List<BlobMetadata>> listener) {
         blobStore.blobContainer((BlobPath) path).listBlobsByPrefixInSortedOrder("", limit, LEXICOGRAPHIC, listener);
     }
 
@@ -143,13 +143,7 @@ public void listAllInSortedOrderAsync(
         int limit,
         ActionListener<List<BlobMetadata>> listener
     ) {
-        threadPool.executor(threadpoolName).execute(() -> {
-            try {
-                listAllInSortedOrder(path, limit, listener);
-            } catch (IOException e) {
-                listener.onFailure(e);
-            }
-        });
+        threadPool.executor(threadpoolName).execute(() -> { listAllInSortedOrder(path, limit, listener); });
     }
 
 }
diff --git a/server/src/main/java/org/opensearch/index/translog/transfer/TransferService.java b/server/src/main/java/org/opensearch/index/translog/transfer/TransferService.java
@@ -107,7 +107,7 @@ void uploadBlobAsync(
      */
     InputStream downloadBlob(Iterable<String> path, String fileName) throws IOException;
 
-    void listAllInSortedOrder(Iterable<String> path, int limit, ActionListener<List<BlobMetadata>> listener) throws IOException;
+    void listAllInSortedOrder(Iterable<String> path, int limit, ActionListener<List<BlobMetadata>> listener);
 
     void listAllInSortedOrderAsync(String threadpoolName, Iterable<String> path, int limit, ActionListener<List<BlobMetadata>> listener);
 

diff --git a/server/src/main/java/org/opensearch/index/translog/transfer/TranslogTransferManager.java b/server/src/main/java/org/opensearch/index/translog/transfer/TranslogTransferManager.java
@@ -209,7 +209,7 @@ public TranslogTransferMetadata readMetadata() throws IOException {
         try {
             transferService.listAllInSortedOrder(remoteMetadataTransferPath, 1, latchedActionListener);
             latch.await();
-        } catch (InterruptedException | IOException e) {
+        } catch (InterruptedException e) {
             throw new IOException("Exception while reading/downloading metadafile", e);
         }
 

diff --git a/...er/src/test/java/org/opensearch/index/translog/transfer/TranslogTransferManagerTests.java b/...er/src/test/java/org/opensearch/index/translog/transfer/TranslogTransferManagerTests.java
@@ -201,7 +201,8 @@ public void testReadMetadataNoFile() throws IOException {
         assertNull(translogTransferManager.readMetadata());
     }
 
-    public void testReadMetadataHappy() throws IOException {
+    // This should happen most of the time - Just a single metadata file
+    public void testReadMetadataSingleFile() throws IOException {
         TranslogTransferManager translogTransferManager = new TranslogTransferManager(
             shardId,
             transferService,
@@ -391,6 +392,43 @@ public void testDeleteTranslogSuccess() throws Exception {
         verify(blobContainer).deleteBlobsIgnoringIfNotExists(eq(files));
     }
 
+    public void testDeleteStaleTranslogMetadata() {
+        TranslogTransferManager translogTransferManager = new TranslogTransferManager(
+            shardId,
+            transferService,
+            remoteBaseTransferPath,
+            null
+        );
+        String tm1 = new TranslogTransferMetadata(1, 1, 1, 2).getFileName();
+        String tm2 = new TranslogTransferMetadata(1, 2, 1, 2).getFileName();
+        String tm3 = new TranslogTransferMetadata(2, 3, 1, 2).getFileName();
+        doAnswer(invocation -> {
+            ActionListener<List<BlobMetadata>> actionListener = invocation.getArgument(3);
+            List<BlobMetadata> bmList = new LinkedList<>();
+            bmList.add(new PlainBlobMetadata(tm1, 1));
+            bmList.add(new PlainBlobMetadata(tm2, 1));
+            bmList.add(new PlainBlobMetadata(tm3, 1));
+            actionListener.onResponse(bmList);
+            return null;
+        }).when(transferService)
+            .listAllInSortedOrderAsync(eq(ThreadPool.Names.REMOTE_PURGE), any(BlobPath.class), anyInt(), any(ActionListener.class));
+        List<String> files = List.of(tm2, tm3);
+        translogTransferManager.deleteStaleTranslogMetadataFilesAsync(() -> {
+            verify(transferService).listAllInSortedOrderAsync(
+                eq(ThreadPool.Names.REMOTE_PURGE),
+                any(BlobPath.class),
+                eq(Integer.MAX_VALUE),
+                any()
+            );
+            verify(transferService).deleteBlobsAsync(
+                eq(ThreadPool.Names.REMOTE_PURGE),
+                any(BlobPath.class),
+                eq(files),
+                any(ActionListener.class)
+            );
+        });
+    }
+
     public void testDeleteTranslogFailure() throws Exception {
         FileTransferTracker tracker = new FileTransferTracker(new ShardId("index", "indexUuid", 0));
         BlobStore blobStore = mock(BlobStore.class);