Part 1 of #58 - fixing state mgmt

specmesh · Oct 2, 2023 · 5763b8f · 5763b8f
1 parent ee689ca
commit 5763b8f
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 1 deletion.
diff --git a/kafka/src/main/java/io/specmesh/kafka/provision/TopicMutators.java b/kafka/src/main/java/io/specmesh/kafka/provision/TopicMutators.java
@@ -237,14 +237,21 @@ public Collection<Topic> mutate(final Collection<Topic> topics)
                             .collect(Collectors.toList());
 
             try {
+                unwanted.forEach(topic -> topic.state(STATE.DELETE));
                 if (!dryRun) {
                     adminClient
                             .deleteTopics(toTopicNames(unwanted))
                             .all()
                             .get(Provisioner.REQUEST_TIMEOUT, TimeUnit.SECONDS);
+                    unwanted.forEach(topic -> topic.state(STATE.DELETED));
                 }
             } catch (InterruptedException | ExecutionException | TimeoutException ex) {
-                throw new ProvisioningException("Failed to cleanup unwanted topics", ex);
+                unwanted.forEach(
+                        topic ->
+                                topic.exception(
+                                                new ProvisioningException(
+                                                        "failed to delete topics", ex))
+                                        .state(STATE.FAILED));
             }
             return unwanted;
         }

diff --git a/kafka/src/test/java/io/specmesh/kafka/ProvisionerUpdatingFunctionalTest.java b/kafka/src/test/java/io/specmesh/kafka/ProvisionerUpdatingFunctionalTest.java
@@ -234,6 +234,7 @@ void shouldCleanupNonSpecTopicsDryRun()
                     TopicProvisioner.provision(true, true, API_SPEC, adminClient);
             // 'should.not.be' topic that should not be
             assertThat(unSpecifiedTopics, is(hasSize(1)));
+            assertThat(unSpecifiedTopics.iterator().next().state(), is(STATE.DELETE));
             assertThat(topicCount(adminClient), is(3L));
         }
     }
@@ -251,6 +252,7 @@ void shouldCleanupNonSpecTopicsIRL() throws ExecutionException, InterruptedExcep
 
             // 'should.not.be' topic that should not be
             assertThat(unSpecifiedTopics, is(hasSize(1)));
+            assertThat(unSpecifiedTopics.iterator().next().state(), is(STATE.DELETED));
 
             // 'should.not.be' topic was removed
             assertThat(topicCount(adminClient), is(2L));