fix replication initialization from newly created backup create many backups

Author: teoincontatto

stackgres-k8s/src/operator/src/main/java/io/stackgres/operator/conciliation/cluster/ClusterRequiredResourcesGenerator.java

@@ -441,30 +441,44 @@ private Optional<StackGresBackup> getReplicationInitializationBackupToCreate(
             labelFactory.replicationInitializationBackupLabels(cluster)))
         .filter(backup -> backup.getSpec().getSgCluster().equals(
             cluster.getMetadata().getName()))
-        .filter(backup -> Optional.ofNullable(backup.getStatus())
-            .map(StackGresBackupStatus::getSgBackupConfig)
-            .map(StackGresBackupConfigSpec::getStorage)
-            .equals(backupObjectStorage.map(StackGresObjectStorage::getSpec)))
-        .filter(backup -> Optional.ofNullable(backup.getStatus())
-            .map(StackGresBackupStatus::getBackupPath)
-            .equals(Optional
-                .ofNullable(cluster.getSpec().getConfigurations().getBackups())
-                .map(Collection::stream)
-                .flatMap(Stream::findFirst)
-                .map(StackGresClusterBackupConfiguration::getPath)))
-        .filter(backup -> Optional.ofNullable(backup.getStatus())
+        .filter(backup -> backup.getStatus() == null
+            || Optional.ofNullable(backup.getStatus())
+            .filter(status -> Optional.of(status)
+                .map(StackGresBackupStatus::getProcess)
+                .map(StackGresBackupProcess::getStatus)
+                .filter(Predicate.not(BackupStatus.COMPLETED.toString()::equals)
+                    .and(Predicate.not(BackupStatus.FAILED.toString()::equals)))
+                .isPresent())
+            .isPresent()
+            || Optional.ofNullable(backup.getStatus())
+            .filter(status -> Optional.of(status)
+                .map(StackGresBackupStatus::getSgBackupConfig)
+                .map(StackGresBackupConfigSpec::getStorage)
+                .equals(backupObjectStorage.map(StackGresObjectStorage::getSpec)))
+            .filter(status -> Optional.of(status)
+                .map(StackGresBackupStatus::getBackupPath)
+                .equals(Optional
+                    .ofNullable(cluster.getSpec().getConfigurations().getBackups())
+                    .map(Collection::stream)
+                    .flatMap(Stream::findFirst)
+                    .map(StackGresClusterBackupConfiguration::getPath)))
+            .filter(status -> Optional.of(status)
+                .map(StackGresBackupStatus::getBackupInformation)
+                .map(StackGresBackupInformation::getPostgresMajorVersion)
+                .filter(postgresMajorVersion::equals)
+                .isPresent())
+            .filter(status -> Optional.of(status)
+                .map(StackGresBackupStatus::getProcess)
+                .map(StackGresBackupProcess::getStatus)
+                .filter(BackupStatus.COMPLETED.toString()::equals)
+                .isPresent())
             .map(StackGresBackupStatus::getProcess)
             .map(StackGresBackupProcess::getTiming)
             .map(StackGresBackupTiming::getEnd)
             .map(Instant::parse)
             .or(() -> Optional.of(now))
             .filter(end -> backupNewerThan.map(end::isAfter).orElse(true))
             .isPresent())
-        .filter(backup -> Optional.ofNullable(backup.getStatus())
-            .map(StackGresBackupStatus::getBackupInformation)
-            .map(StackGresBackupInformation::getPostgresMajorVersion)
-            .filter(postgresMajorVersion::equals)
-            .isPresent())
         .findFirst();
   }