enhancement - #58 topic cleanup non-spec topic - part 1 (#199)

* #58 Topics: remove topics that aren't in the spec.
note on 'Provision' command updated flags:
- use -dry or --dry-run to check action (inc cleaning)
- use -clean or --clean-unspecified to clean resources

* spotless

* Part 1 of #58 - fixing state mgmt

* enhancement - #58 schema cleanup of non-spec - part 2 (#210)

* Part 2 of #58 - cleanup unspecified schemas

* spotless

* fixed compilation

cli/src/main/java/io/specmesh/cli/Provision.java

@@ -90,13 +90,22 @@ public static void main(final String[] args) {
     @Option(
             names = {"-dry", "--dry-run"},
             description =
-                    "Compares the cluster against the spec, outputting proposed changes if"
-                        + " compatible.If the spec incompatible with the cluster (not sure how it"
-                        + " could be) then will fail with a descriptive error message.A return"
-                        + " value of 0=indicates no changes needed; 1=changes needed; -1=not"
-                        + " compatible, blah blah")
+                    "Compares the cluster resources against the spec, outputting proposed changes"
+                        + " if compatible. If the spec incompatible with the cluster (not sure how"
+                        + " it could be) then will fail with a descriptive error message. A return"
+                        + " value of '0' = indicates no changes needed; '1' = changes needed; '-1'"
+                        + " not compatible")
     private boolean dryRun;
 
+    @Option(
+            names = {"-clean", "--clean-unspecified"},
+            description =
+                    "Compares the cluster resources against the spec, outputting proposed set of"
+                        + " resources that are unexpected (not specified). Use with '-dry-run' for"
+                        + " non-destructive checks. This operation will not create resources, it"
+                        + " will only remove unspecified resources")
+    private boolean cleanUnspecified;
+
     @Option(
             names = "-D",
             mapFallbackValue = "",
@@ -113,6 +122,7 @@ public Integer call() throws Exception {
         final var status =
                 Provisioner.provision(
                         dryRun,
+                        cleanUnspecified,
                         specMeshSpec(),
                         schemaPath,
                         Clients.adminClient(brokerUrl, username, secret),

cli/src/test/java/io/specmesh/cli/StorageConsumptionFunctionalTest.java

@@ -75,6 +75,7 @@ class StorageConsumptionFunctionalTest {
     void shouldGetStorageAndConsumptionMetrics() throws Exception {
 
         Provisioner.provision(
+                false,
                 false,
                 API_SPEC,
                 "./build/resources/test",

kafka-test/src/test/java/io/specmesh/kafka/ClientsFunctionalDemoTest.java

@@ -107,6 +107,7 @@ public static void provision() {
             final SchemaRegistryClient schemaRegistryClient =
                     new CachedSchemaRegistryClient(KAFKA_ENV.schemeRegistryServer(), 5);
             Provisioner.provision(
+                    false,
                     false,
                     API_SPEC,
                     "./build/resources/test",

kafka/src/main/java/io/specmesh/kafka/provision/Provisioner.java

@@ -32,6 +32,7 @@ private Provisioner() {}
      * Provision Topics, ACLS and schemas
      *
      * @param dryRun test or execute
+     * @param cleanUnspecified cleanup
      * @param apiSpec given spec
      * @param schemaResources schema path
      * @param adminClient kafka admin client
@@ -41,6 +42,7 @@ private Provisioner() {}
      */
     public static Status provision(
             final boolean dryRun,
+            final boolean cleanUnspecified,
             final KafkaApiSpec apiSpec,
             final String schemaResources,
             final Admin adminClient,
@@ -49,12 +51,19 @@ public static Status provision(
         apiSpec.apiSpec().validate();
 
         final var status =
-                Status.builder().topics(TopicProvisioner.provision(dryRun, apiSpec, adminClient));
+                Status.builder()
+                        .topics(
+                                TopicProvisioner.provision(
+                                        dryRun, cleanUnspecified, apiSpec, adminClient));
         schemaRegistryClient.ifPresent(
                 registryClient ->
                         status.schemas(
                                 SchemaProvisioner.provision(
-                                        dryRun, apiSpec, schemaResources, registryClient)));
+                                        dryRun,
+                                        cleanUnspecified,
+                                        apiSpec,
+                                        schemaResources,
+                                        registryClient)));
         status.acls(AclProvisioner.provision(dryRun, apiSpec, adminClient));
         return status.build();
     }

kafka/src/main/java/io/specmesh/kafka/provision/SchemaChangeSetCalculators.java

@@ -67,6 +67,24 @@ public Collection<Schema> calculate(
         }
     }
 
+    /** Return set of 'unspecific' (i.e. non-required) schemas */
+    public static class CleanUnspecifiedCalculator implements ChangeSetCalculator {
+
+        /**
+         * remove the required items from the existing.. the remainder are not specified
+         *
+         * @param existing - existing
+         * @param required - needed
+         * @return schemas that aren't specd
+         */
+        @Override
+        public Collection<Schema> calculate(
+                final Collection<Schema> existing, final Collection<Schema> required) {
+            existing.removeAll(required);
+            return existing;
+        }
+    }
+
     /** Returns those schemas to create and ignores existing */
     public static final class UpdateCalculator implements ChangeSetCalculator {
 
@@ -190,11 +208,18 @@ public static ChangeSetBuilder builder() {
         /**
          * build it
          *
+         * @param cleanUnspecified - cleanup
          * @param client sr client
          * @return required calculator
          */
-        public ChangeSetCalculator build(final SchemaRegistryClient client) {
-            return new Collective(new UpdateCalculator(client), new CreateCalculator());
+        public ChangeSetCalculator build(
+                final boolean cleanUnspecified, final SchemaRegistryClient client) {
+            if (cleanUnspecified) {
+                return new CleanUnspecifiedCalculator();
+
+            } else {
+                return new Collective(new UpdateCalculator(client), new CreateCalculator());
+            }
         }
     }
 }