do not remove subscription if it already existed before subscription …

…creation (#1887)

* do not remove subscription if it already existed before subscription creation

* fix MultiDatacenterRepositoryCommandExecutorTest

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/blacklist/commands/AddTopicToBlacklistRepositoryCommand.java

@@ -21,7 +21,7 @@ public void execute(DatacenterBoundRepositoryHolder<TopicBlacklistRepository> ho
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<TopicBlacklistRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<TopicBlacklistRepository> holder, Exception exception) {
         holder.getRepository().remove(qualifiedTopicName);
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/blacklist/commands/RemoveTopicFromBlacklistRepositoryCommand.java

@@ -23,7 +23,7 @@ public void execute(DatacenterBoundRepositoryHolder<TopicBlacklistRepository> ho
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<TopicBlacklistRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<TopicBlacklistRepository> holder, Exception exception) {
         if (exists) {
             holder.getRepository().add(qualifiedTopicName);
         }

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/credentials/commands/UpdateCredentialsRepositoryCommand.java

@@ -24,7 +24,7 @@ public void execute(DatacenterBoundRepositoryHolder<CredentialsRepository> holde
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<CredentialsRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<CredentialsRepository> holder, Exception exception) {
     }
 
     @Override

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/dc/MultiDatacenterRepositoryCommandExecutor.java

@@ -55,27 +55,27 @@ private <T> void execute(RepositoryCommand<T> command, boolean isRollbackEnabled
             } catch (RepositoryNotAvailableException e) {
                 logger.warn("Execute failed with an RepositoryNotAvailableException error", e);
                 if (isRollbackEnabled) {
-                    rollback(executedRepoHolders, command);
+                    rollback(executedRepoHolders, command, e);
                 }
                 if (shouldStopExecutionOnFailure) {
                     throw ExceptionWrapper.wrapInInternalProcessingExceptionIfNeeded(e, command.toString(), repoHolder.getDatacenterName());
                 }
             } catch (Exception e) {
                 logger.warn("Failed to execute repository command: {} in ZK dc: {} in: {} ms", command, repoHolder.getDatacenterName(), System.currentTimeMillis() - start, e);
                 if (isRollbackEnabled) {
-                    rollback(executedRepoHolders, command);
+                    rollback(executedRepoHolders, command, e);
                 }
                 throw ExceptionWrapper.wrapInInternalProcessingExceptionIfNeeded(e, command.toString(), repoHolder.getDatacenterName());
             }
         }
     }
 
-    private <T> void rollback(List<DatacenterBoundRepositoryHolder<T>> repoHolders, RepositoryCommand<T> command) {
+    private <T> void rollback(List<DatacenterBoundRepositoryHolder<T>> repoHolders, RepositoryCommand<T> command, Exception exception) {
         long start = System.currentTimeMillis();
         for (DatacenterBoundRepositoryHolder<T> repoHolder : repoHolders) {
             logger.info("Executing rollback of repository command: {} in ZK dc: {}", command, repoHolder.getDatacenterName());
             try {
-                command.rollback(repoHolder);
+                command.rollback(repoHolder, exception);
                 logger.info("Successfully executed rollback of repository command: {} in ZK dc: {} in: {} ms", command, repoHolder.getDatacenterName(), System.currentTimeMillis() - start);
             } catch (Exception e) {
                 logger.error("Rollback procedure failed for command {} on DC {}", command, repoHolder.getDatacenterName(), e);

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/dc/RepositoryCommand.java

@@ -6,7 +6,7 @@ public abstract class RepositoryCommand<T> {
 
     public abstract void execute(DatacenterBoundRepositoryHolder<T> holder);
 
-    public abstract void rollback(DatacenterBoundRepositoryHolder<T> holder);
+    public abstract void rollback(DatacenterBoundRepositoryHolder<T> holder, Exception exception);
 
     public abstract Class<T> getRepositoryType();
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/group/commands/CreateGroupRepositoryCommand.java

@@ -25,7 +25,7 @@ public void execute(DatacenterBoundRepositoryHolder<GroupRepository> holder) {
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<GroupRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<GroupRepository> holder, Exception exception) {
         if (!exists) {
             holder.getRepository().removeGroup(group.getGroupName());
         }

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/group/commands/RemoveGroupRepositoryCommand.java

@@ -26,7 +26,7 @@ public void execute(DatacenterBoundRepositoryHolder<GroupRepository> holder) {
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<GroupRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<GroupRepository> holder, Exception exception) {
         holder.getRepository().createGroup(backup);
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/group/commands/UpdateGroupRepositoryCommand.java

@@ -26,7 +26,7 @@ public void execute(DatacenterBoundRepositoryHolder<GroupRepository> holder) {
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<GroupRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<GroupRepository> holder, Exception exception) {
         holder.getRepository().updateGroup(backup);
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/oauth/commands/CreateOAuthProviderRepositoryCommand.java

@@ -22,7 +22,7 @@ public void execute(DatacenterBoundRepositoryHolder<OAuthProviderRepository> hol
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<OAuthProviderRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<OAuthProviderRepository> holder, Exception exception) {
         holder.getRepository().removeOAuthProvider(provider.getName());
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/oauth/commands/RemoveOAuthProviderRepositoryCommand.java

@@ -26,7 +26,7 @@ public void execute(DatacenterBoundRepositoryHolder<OAuthProviderRepository> hol
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<OAuthProviderRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<OAuthProviderRepository> holder, Exception exception) {
         holder.getRepository().createOAuthProvider(backup);
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/oauth/commands/UpdateOAuthProviderRepositoryCommand.java

@@ -26,7 +26,7 @@ public void execute(DatacenterBoundRepositoryHolder<OAuthProviderRepository> hol
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<OAuthProviderRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<OAuthProviderRepository> holder, Exception exception) {
         holder.getRepository().updateOAuthProvider(backup);
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/readiness/SetReadinessCommand.java

@@ -22,7 +22,7 @@ public void execute(DatacenterBoundRepositoryHolder<DatacenterReadinessRepositor
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<DatacenterReadinessRepository> holder) { }
+    public void rollback(DatacenterBoundRepositoryHolder<DatacenterReadinessRepository> holder, Exception exception) { }
 
     @Override
     public Class<DatacenterReadinessRepository> getRepositoryType() {

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/retransmit/CreateOfflineRetransmissionTaskCommand.java

@@ -21,7 +21,7 @@ public void execute(DatacenterBoundRepositoryHolder<OfflineRetransmissionReposit
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<OfflineRetransmissionRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<OfflineRetransmissionRepository> holder, Exception exception) {
         holder.getRepository().deleteTask(task.getTaskId());
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/retransmit/DeleteOfflineRetransmissionTaskCommand.java

@@ -21,7 +21,7 @@ public void execute(DatacenterBoundRepositoryHolder<OfflineRetransmissionReposit
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<OfflineRetransmissionRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<OfflineRetransmissionRepository> holder, Exception exception) {
 
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/retransmit/RetransmitCommand.java

@@ -23,7 +23,7 @@ public void execute(DatacenterBoundRepositoryHolder<AdminTool> holder) {
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<AdminTool> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<AdminTool> holder, Exception exception) {
 
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/subscription/commands/CreateSubscriptionRepositoryCommand.java

@@ -1,6 +1,7 @@
 package pl.allegro.tech.hermes.management.domain.subscription.commands;
 
 import pl.allegro.tech.hermes.api.Subscription;
+import pl.allegro.tech.hermes.domain.subscription.SubscriptionAlreadyExistsException;
 import pl.allegro.tech.hermes.domain.subscription.SubscriptionRepository;
 import pl.allegro.tech.hermes.management.domain.dc.DatacenterBoundRepositoryHolder;
 import pl.allegro.tech.hermes.management.domain.dc.RepositoryCommand;
@@ -22,7 +23,11 @@ public void execute(DatacenterBoundRepositoryHolder<SubscriptionRepository> hold
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<SubscriptionRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<SubscriptionRepository> holder, Exception exception) {
+        if (exception instanceof SubscriptionAlreadyExistsException) {
+            // prevents removal of already existing subscription
+            return;
+        }
         holder.getRepository().removeSubscription(subscription.getTopicName(), subscription.getName());
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/subscription/commands/RemoveSubscriptionRepositoryCommand.java

@@ -30,7 +30,7 @@ public void execute(DatacenterBoundRepositoryHolder<SubscriptionRepository> hold
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<SubscriptionRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<SubscriptionRepository> holder, Exception exception) {
         holder.getRepository().createSubscription(backup);
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/subscription/commands/UpdateSubscriptionRepositoryCommand.java

@@ -25,7 +25,7 @@ public void execute(DatacenterBoundRepositoryHolder<SubscriptionRepository> hold
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<SubscriptionRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<SubscriptionRepository> holder, Exception exception) {
         holder.getRepository().updateSubscription(backup);
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/topic/commands/CreateTopicRepositoryCommand.java

@@ -22,7 +22,7 @@ public void execute(DatacenterBoundRepositoryHolder<TopicRepository> holder) {
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<TopicRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<TopicRepository> holder, Exception exception) {
         /*
         We don't want to do a rollback due to possible race conditions with creating a topic on Kafka.
         It increases the probability of discrepancies between Kafka and Zookeeper: topic exists in Kafka,

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/topic/commands/RemoveTopicRepositoryCommand.java

@@ -22,7 +22,7 @@ public void execute(DatacenterBoundRepositoryHolder<TopicRepository> holder) {
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<TopicRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<TopicRepository> holder, Exception exception) {
         /*
         We don't want to do a rollback due to possible race conditions with creating a topic on Kafka.
         It increases the probability of discrepancies between Kafka and Zookeeper: topic exists in Kafka,

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/topic/commands/TouchTopicRepositoryCommand.java

@@ -22,7 +22,7 @@ public void execute(DatacenterBoundRepositoryHolder<TopicRepository> holder) {
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<TopicRepository> holder) {}
+    public void rollback(DatacenterBoundRepositoryHolder<TopicRepository> holder, Exception exception) {}
 
     @Override
     public Class<TopicRepository> getRepositoryType() {

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/topic/commands/UpdateTopicRepositoryCommand.java

@@ -26,7 +26,7 @@ public void execute(DatacenterBoundRepositoryHolder<TopicRepository> holder) {
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<TopicRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<TopicRepository> holder, Exception exception) {
         holder.getRepository().updateTopic(backup);
     }
 

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/workload/constraints/command/CreateSubscriptionConstraintsRepositoryCommand.java

@@ -28,7 +28,7 @@ public void execute(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepositor
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder, Exception exception) {
         if (!exists) {
             holder.getRepository().deleteConstraints(subscriptionName);
         }

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/workload/constraints/command/CreateTopicConstraintsRepositoryCommand.java

@@ -28,7 +28,7 @@ public void execute(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepositor
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder, Exception exception) {
         if (!exist) {
             holder.getRepository().deleteConstraints(topicName);
         }

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/workload/constraints/command/DeleteSubscriptionConstraintsRepositoryCommand.java

@@ -26,7 +26,7 @@ public void execute(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepositor
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder, Exception exception) {
         if (backup != null) {
             holder.getRepository().createConstraints(subscriptionName, backup);
         }

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/workload/constraints/command/DeleteTopicConstraintsRepositoryCommand.java

@@ -26,7 +26,7 @@ public void execute(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepositor
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder, Exception exception) {
         if (backup != null) {
             holder.getRepository().createConstraints(topicName, backup);
         }

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/workload/constraints/command/UpdateSubscriptionConstraintsRepositoryCommand.java

@@ -28,7 +28,7 @@ public void execute(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepositor
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder, Exception exception) {
         if (backup != null) {
             holder.getRepository().updateConstraints(subscriptionName, backup);
         }

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/workload/constraints/command/UpdateTopicConstraintsRepositoryCommand.java

@@ -28,7 +28,7 @@ public void execute(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepositor
     }
 
     @Override
-    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder) {
+    public void rollback(DatacenterBoundRepositoryHolder<WorkloadConstraintsRepository> holder, Exception exception) {
         if (backup != null) {
             holder.getRepository().updateConstraints(topicName, backup);
         }

hermes-management/src/test/groovy/pl/allegro/tech/hermes/management/domain/dc/MultiDatacenterRepositoryCommandExecutorTest.groovy

@@ -66,7 +66,7 @@ class MultiDatacenterRepositoryCommandExecutorTest extends Specification {
         executor.execute(command)
 
         then:
-        1 * command.rollback(holder1)
+        1 * command.rollback(holder1, _)
 
         thrown InternalProcessingException
     }
@@ -104,7 +104,7 @@ class MultiDatacenterRepositoryCommandExecutorTest extends Specification {
         executor.executeByUser(command, ADMIN)
 
         then:
-        1 * command.rollback(holder1)
+        1 * command.rollback(holder1, _)
 
         thrown InternalProcessingException
     }
@@ -160,7 +160,7 @@ class MultiDatacenterRepositoryCommandExecutorTest extends Specification {
         executor.executeByUser(command, NON_ADMIN)
 
         then:
-        1 * command.rollback(holder1)
+        1 * command.rollback(holder1, _)
 
         thrown InternalProcessingException
     }

hermes-management/src/test/groovy/pl/allegro/tech/hermes/management/domain/subscription/commands/CreateSubscriptionRepositoryCommandTest.groovy

@@ -0,0 +1,53 @@
+package pl.allegro.tech.hermes.management.domain.subscription.commands
+
+import pl.allegro.tech.hermes.api.Subscription
+import pl.allegro.tech.hermes.api.TopicName
+import pl.allegro.tech.hermes.domain.subscription.SubscriptionAlreadyExistsException
+import pl.allegro.tech.hermes.domain.subscription.SubscriptionRepository
+import pl.allegro.tech.hermes.management.domain.dc.DatacenterBoundRepositoryHolder
+import spock.lang.Shared
+import spock.lang.Specification
+
+import static pl.allegro.tech.hermes.test.helper.builder.SubscriptionBuilder.subscription
+
+class CreateSubscriptionRepositoryCommandTest extends Specification {
+
+    @Shared
+    def topicName = new TopicName("group", "topic")
+
+    @Shared
+    def subscriptionName = "subscription"
+
+    @Shared
+    Subscription subscription = subscription(topicName, subscriptionName).build()
+
+    def "should not remove subscription if subscription already exists during rollback"() {
+        given:
+        SubscriptionRepository subscriptionRepository = Mock(SubscriptionRepository)
+        DatacenterBoundRepositoryHolder<SubscriptionRepository> repository = Mock(DatacenterBoundRepositoryHolder) {
+            getRepository() >> subscriptionRepository
+        }
+        def command = new CreateSubscriptionRepositoryCommand(subscription)
+
+        when:
+        command.rollback(repository, new SubscriptionAlreadyExistsException(subscription))
+
+        then:
+        0 * subscriptionRepository.removeSubscription(topicName, subscriptionName)
+    }
+
+    def "should remove subscription during rollback"() {
+        given:
+        SubscriptionRepository subscriptionRepository = Mock(SubscriptionRepository)
+        DatacenterBoundRepositoryHolder<SubscriptionRepository> repository = Mock(DatacenterBoundRepositoryHolder) {
+            getRepository() >> subscriptionRepository
+        }
+        def command = new CreateSubscriptionRepositoryCommand(subscription)
+
+        when:
+        command.rollback(repository, new RuntimeException())
+
+        then:
+        1 * subscriptionRepository.removeSubscription(topicName, subscriptionName)
+    }
+}