merge: #10213

10213: feat(backup): backup manager can query status r=oleschoenburg a=deepthidevaki ## Description - Backup manager can query the status of backup in the given backup store ## Related issues closes #9981 Co-authored-by: Deepthi Devaki Akkoorath <deepthidevaki@gmail.com>
camunda · Aug 30, 2022 · 1480f7e · 1480f7e
2 parents bbd0e5d + 21a5c37
commit 1480f7e
Show file tree

Hide file tree

Showing 3 changed files with 60 additions and 3 deletions.
diff --git a/backup/src/main/java/io/camunda/zeebe/backup/management/BackupService.java b/backup/src/main/java/io/camunda/zeebe/backup/management/BackupService.java
@@ -85,7 +85,7 @@ public void takeBackup(final long checkpointId, final long checkpointPosition) {
           final InProgressBackupImpl inProgressBackup =
               new InProgressBackupImpl(
                   snapshotStore,
-                  new BackupIdentifierImpl(nodeId, partitionId, checkpointId),
+                  getBackupId(checkpointId),
                   checkpointPosition,
                   numberOfPartitions,
                   actor,
@@ -113,13 +113,16 @@ public void takeBackup(final long checkpointId, final long checkpointPosition) {
 
   @Override
   public ActorFuture<BackupStatus> getBackupStatus(final long checkpointId) {
-    return CompletableActorFuture.completedExceptionally(
-        new UnsupportedOperationException("Not implemented"));
+    return internalBackupManager.getBackupStatus(getBackupId(checkpointId), actor);
   }
 
   @Override
   public ActorFuture<Void> deleteBackup(final long checkpointId) {
     return CompletableActorFuture.completedExceptionally(
         new UnsupportedOperationException("Not implemented"));
   }
+
+  private BackupIdentifierImpl getBackupId(final long checkpointId) {
+    return new BackupIdentifierImpl(nodeId, partitionId, checkpointId);
+  }
 }
diff --git a/backup/src/main/java/io/camunda/zeebe/backup/management/BackupServiceImpl.java b/backup/src/main/java/io/camunda/zeebe/backup/management/BackupServiceImpl.java
@@ -7,9 +7,12 @@
  */
 package io.camunda.zeebe.backup.management;
 
+import io.camunda.zeebe.backup.api.BackupIdentifier;
+import io.camunda.zeebe.backup.api.BackupStatus;
 import io.camunda.zeebe.backup.api.BackupStore;
 import io.camunda.zeebe.scheduler.ConcurrencyControl;
 import io.camunda.zeebe.scheduler.future.ActorFuture;
+import io.camunda.zeebe.scheduler.future.CompletableActorFuture;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.function.BiConsumer;
@@ -117,4 +120,22 @@ private BiConsumer<Void, Throwable> proceed(
       }
     };
   }
+
+  ActorFuture<BackupStatus> getBackupStatus(
+      final BackupIdentifier backupId, final ConcurrencyControl executor) {
+    final var future = new CompletableActorFuture<BackupStatus>();
+    executor.run(
+        () ->
+            backupStore
+                .getStatus(backupId)
+                .whenComplete(
+                    (status, error) -> {
+                      if (error == null) {
+                        future.complete(status);
+                      } else {
+                        future.completeExceptionally(error);
+                      }
+                    }));
+    return future;
+  }
 }
diff --git a/backup/src/test/java/io/camunda/zeebe/backup/management/BackupServiceImplTest.java b/backup/src/test/java/io/camunda/zeebe/backup/management/BackupServiceImplTest.java
@@ -13,7 +13,9 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import io.camunda.zeebe.backup.api.BackupStatus;
 import io.camunda.zeebe.backup.api.BackupStore;
+import io.camunda.zeebe.backup.common.BackupIdentifierImpl;
 import io.camunda.zeebe.scheduler.ConcurrencyControl;
 import io.camunda.zeebe.scheduler.future.ActorFuture;
 import io.camunda.zeebe.scheduler.testing.TestConcurrencyControl;
@@ -163,6 +165,37 @@ void shouldFailBackupIfStoringFailed() {
     verifyInProgressBackupIsCleanedUpAfterFailure();
   }
 
+  @Test
+  void shouldGetBackupStatus() {
+    // given
+    final BackupStatus status = mock(BackupStatus.class);
+    when(backupStore.getStatus(any())).thenReturn(CompletableFuture.completedFuture(status));
+
+    // when
+    final var result =
+        backupService.getBackupStatus(new BackupIdentifierImpl(1, 1, 1), concurrencyControl);
+
+    // then
+    assertThat(result).succeedsWithin(Duration.ofMillis(100)).isEqualTo(status);
+  }
+
+  @Test
+  void shouldCompleteFutureWhenBackupStatusFailed() {
+    // given
+    when(backupStore.getStatus(any()))
+        .thenReturn(CompletableFuture.failedFuture(new RuntimeException("Expected")));
+
+    // when
+    final var result =
+        backupService.getBackupStatus(new BackupIdentifierImpl(1, 1, 1), concurrencyControl);
+
+    // then
+    assertThat(result)
+        .failsWithin(Duration.ofMillis(100))
+        .withThrowableOfType(ExecutionException.class)
+        .withMessageContaining("Expected");
+  }
+
   private ActorFuture<Void> failedFuture() {
     final ActorFuture<Void> future = concurrencyControl.createFuture();
     future.completeExceptionally(new RuntimeException("Expected"));