Repository: kafka-ops/kafka-topology-builder (java)

Issue:
Allow for validators in other packages than "com.purbon.kafka.topology.validation"
**Is your feature request related to a problem? Please describe.**
The validation feature is nice. But it should support creating your own custom validators in a better way. As it is now, it is kind of limited to validators shipping with KTB as all validators must be placed in the package "com.purbon.kafka.topology.validation".

**Describe the solution you'd like**
The config `topology.validations` should allow for specifying fully qualified class names. 

**Describe alternatives you've considered**
Today you either have to commit the validator to KTB code base, which is fine if it's a generic validation that others can re-use, or create your own outside of KTB but use the required package structure.

**Additional context**
The documentation on `topology.validations` config is not explicit about this package/naming convention, so if you don't read the code it is a bit confusing. I think it would be a good thing to change also the naming for KTB built-in validator. Either one of these options:
- Use the fully qualified class names also for any KTB built-in validators. I think this is the most clean solution. 
- Or just use the class name when it's a KTB validator. KTB should then look for the class in "com.purbon.kafka.topology.validation". Either by moving the validators up one package or by looking for the class in sub packages.


Pull request description:
Support topology validations with fully qualified class names as configuration values (backwards compatible)
This fixes #201. 

In addition the following improvements have been done:
- Fail on validation class not found in class path. KTB should not preceed with processing in this case.

* **Please check if the PR fulfills these requirements**
- [x] The commit messages are descriptive
- [x] Tests for the changes have been added (for bug fixes / features)
- [x] Docs have been added / updated (for bug fixes / features)
- [x] An issue has been created for the pull requests. Some issues might require previous discussion.

* **What kind of change does this PR introduce?** (Bug fix, feature, docs update, ...)
Feature improvement.

* **What is the current behavior?** (You can also link to an open issue here)
See #201.

* **What is the new behavior (if this is a feature change)?**
See #201.

* **Does this PR introduce a breaking change?** (What changes might users need to make in their application due to this PR?)
No. Removing old, now deprecated behavior, should be handled in a later commit.


Patch:
diff --git a/docs/config-values.rst b/docs/config-values.rst
index 3290012..21b12f2 100644
--- a/docs/config-values.rst
+++ b/docs/config-values.rst
@@ -104,24 +104,27 @@ An example configuration might look like this:
     kafka.internal.topic.prefixes.1=topicPrefixA
     kafka.internal.topic.prefixes.2=topicPrefixB
 
-Topology level validations
+Topology validations
 -----------
 
-It is now possible to define a list of validations to be applied to the desired Topology file.
+It is possible to define a list of validations to be applied to the topology files.
 
-As a user you can list the validations to be applied using the configuration property:
+As a user you can list the validation classes to be applied using the configuration property:
 
 - **Property**: *topology.validations*
 
-This property accepts the list of validations available in the class path.
+This property accepts a list of validation classes available in the class path. Use the fully qualified class name.
 They will be applied in sequence as defined.
+You will find included KTB validations in the package 'com.purbon.kafka.topology.validation'.
 
 An example configuration might look like this:
 ::
-    topology.validations.0=topology.CamelCaseNameFormatValidation
-    topology.validations.1=topic.PartitionNumberValidation
+    topology.validations.0=com.purbon.kafka.topology.validation.topology.CamelCaseNameFormatValidation
+    topology.validations.1=com.purbon.kafka.topology.validation.topic.PartitionNumberValidation
 
-Users can pull custom validation available from the class path.
+You can also create your own custom validations. The validations must implement one of these interfaces:
+- com.purbon.kafka.topology.validation.TopologyValidation
+- com.purbon.kafka.topology.validation.TopicValidation
 
 Prevent ACL for topic creation for connector principal
 -----------
diff --git a/src/main/java/com/purbon/kafka/topology/TopologyValidator.java b/src/main/java/com/purbon/kafka/topology/TopologyValidator.java
index a84c81f..2913a71 100644
--- a/src/main/java/com/purbon/kafka/topology/TopologyValidator.java
+++ b/src/main/java/com/purbon/kafka/topology/TopologyValidator.java
@@ -76,18 +76,30 @@ public class TopologyValidator {
         .collect(Collectors.toList());
   }
 
-  class VoidValidation implements TopologyValidation {
-    @Override
-    public void valid(Topology topology) {}
-  }
-
   private List<Validation> validations() {
     return config.getTopologyValidations().stream()
         .map(
             validationClass -> {
               try {
-                String fullValidationClass = classPrefix + validationClass;
-                Class<?> clazz = Class.forName(fullValidationClass);
+                Class<?> clazz = getValidationClazz(validationClass);
+                if (clazz == null) {
+                  String deprecatedClassNameFromPrefix = classPrefix + validationClass;
+                  clazz = getValidationClazz(deprecatedClassNameFromPrefix);
+                  if (clazz != null) {
+                    LOGGER.warn(
+                        "Deprecation warning: Specifying validations in the config without using the "
+                            + "fully qualified class name is deprecated. Support will be removed in next major release. "
+                            + "Please switch to using fully qualified class name");
+                  }
+                }
+                if (clazz == null) {
+                  throw new IOException(
+                      String.format(
+                          "Could not find validation class '%s' in class path. "
+                              + "Please use the fully qualified class name and check your config.",
+                          validationClass));
+                }
+
                 Constructor<?> constructor = clazz.getConstructor();
                 Object instance = constructor.newInstance();
                 if (instance instanceof TopologyValidation) {
@@ -95,14 +107,21 @@ public class TopologyValidator {
                 } else if (instance instanceof TopicValidation) {
                   return (TopicValidation) instance;
                 } else {
-                  throw new IOException("invalid validation type " + fullValidationClass);
+                  throw new IOException("invalid validation type specified " + validationClass);
                 }
               } catch (Exception ex) {
-                LOGGER.debug(ex);
-                return new VoidValidation();
+                throw new IllegalStateException(
+                    "Failed to load topology validations from class path", ex);
               }
             })
-        .filter(validation -> !(validation instanceof VoidValidation))
         .collect(Collectors.toList());
   }
+
+  private Class<?> getValidationClazz(String validationClass) {
+    try {
+      return Class.forName(validationClass);
+    } catch (ClassNotFoundException e) {
+      return null;
+    }
+  }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/roles/CCloudAclsProvider.java b/src/main/java/com/purbon/kafka/topology/roles/CCloudAclsProvider.java
index 8d2aef4..f36ac54 100644
--- a/src/main/java/com/purbon/kafka/topology/roles/CCloudAclsProvider.java
+++ b/src/main/java/com/purbon/kafka/topology/roles/CCloudAclsProvider.java
@@ -86,11 +86,12 @@ public class CCloudAclsProvider extends SimpleAclsProvider implements AccessCont
   private TopologyAclBinding convertToConfluentCloudId(
       Map<String, ServiceAccount> serviceAccounts, TopologyAclBinding binding) {
     if (binding.asAclBinding().isPresent()) {
-      AclBinding aclBinding = convertToConfluentCloudId(serviceAccounts, binding.asAclBinding().get());
+      AclBinding aclBinding =
+          convertToConfluentCloudId(serviceAccounts, binding.asAclBinding().get());
       return aclBinding == null ? null : new TopologyAclBinding(aclBinding);
     } else {
       Integer id = getId(serviceAccounts, binding.getPrincipal());
-      return id == null ? null :  getTopologyAclBinding(binding, "User:" + id);
+      return id == null ? null : getTopologyAclBinding(binding, "User:" + id);
     }
   }
 
@@ -119,18 +120,19 @@ public class CCloudAclsProvider extends SimpleAclsProvider implements AccessCont
 
   private AclBinding getAclBinding(AclBinding aclBinding, String principle) {
     AccessControlEntry entry = aclBinding.entry();
-    AccessControlEntry accessControlEntry = new AccessControlEntry(principle, entry.host(), entry.operation(), entry.permissionType());
+    AccessControlEntry accessControlEntry =
+        new AccessControlEntry(principle, entry.host(), entry.operation(), entry.permissionType());
     return new AclBinding(aclBinding.pattern(), accessControlEntry);
   }
 
   private TopologyAclBinding getTopologyAclBinding(TopologyAclBinding binding, String principle) {
     return new TopologyAclBinding(
-            binding.getResourceType(),
-            binding.getResourceName(),
-            binding.getHost(),
-            binding.getOperation(),
-            principle,
-            binding.getPattern());
+        binding.getResourceType(),
+        binding.getResourceName(),
+        binding.getHost(),
+        binding.getOperation(),
+        principle,
+        binding.getPattern());
   }
 
   private Integer getId(Map<String, ServiceAccount> serviceAccounts, String name) {


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java b/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java
index b1d16b6..1da609f 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java
@@ -1,18 +1,18 @@
 package com.purbon.kafka.topology;
 
-import static com.purbon.kafka.topology.BuilderCLI.ADMIN_CLIENT_CONFIG_OPTION;
-import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
-import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPOLOGY_VALIDATIONS_CONFIG;
-import static org.assertj.core.api.Assertions.assertThat;
-
 import com.purbon.kafka.topology.model.Topology;
 import com.purbon.kafka.topology.serdes.TopologySerdes;
 import com.purbon.kafka.topology.utils.TestUtils;
-import java.io.IOException;
-import java.util.*;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.*;
+
+import static com.purbon.kafka.topology.BuilderCLI.ADMIN_CLIENT_CONFIG_OPTION;
+import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
+import static com.purbon.kafka.topology.TopologyBuilderConfig.TOPOLOGY_VALIDATIONS_CONFIG;
+import static org.assertj.core.api.Assertions.assertThat;
+
 public class TopologyValidationTest {
 
   private TopologySerdes parser;
@@ -23,18 +23,14 @@ public class TopologyValidationTest {
   }
 
   @Test
-  public void testPositiveExecutionOnCamelCaseNames() throws IOException {
+  public void testPositiveExecutionOnCamelCaseNames() {
 
     Topology topology =
-        parser.deserialise(TestUtils.getResourceFile("/descriptor-with-camelCaseNames.yml"));
+            parser.deserialise(TestUtils.getResourceFile("/descriptor-with-camelCaseNames.yml"));
 
-    Map<String, String> cliOps = new HashMap<>();
-    cliOps.put(BROKERS_OPTION, "");
-    cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
-
-    Properties props = new Properties();
-    props.put(TOPOLOGY_VALIDATIONS_CONFIG, Arrays.asList("topology.CamelCaseNameFormatValidation"));
-    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    TopologyBuilderConfig config =
+            createTopologyBuilderConfig(
+                    "com.purbon.kafka.topology.validation.topology.CamelCaseNameFormatValidation");
 
     TopologyValidator validator = new TopologyValidator(config);
     List<String> results = validator.validate(topology);
@@ -42,20 +38,15 @@ public class TopologyValidationTest {
   }
 
   @Test
-  public void testInvalidExecutionBecuaseofNumberOfPartitions() throws IOException {
+  public void testInvalidExecutionBecuaseofNumberOfPartitions() {
 
     Topology topology =
         parser.deserialise(TestUtils.getResourceFile("/descriptor-with-camelCaseNames.yml"));
 
-    Map<String, String> cliOps = new HashMap<>();
-    cliOps.put(BROKERS_OPTION, "");
-    cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
-
-    Properties props = new Properties();
-    props.put(
-        TOPOLOGY_VALIDATIONS_CONFIG,
-        Arrays.asList("topology.CamelCaseNameFormatValidation", "topic.PartitionNumberValidation"));
-    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    TopologyBuilderConfig config =
+        createTopologyBuilderConfig(
+                "com.purbon.kafka.topology.validation.topology.CamelCaseNameFormatValidation",
+                "com.purbon.kafka.topology.validation.topic.PartitionNumberValidation");
 
     TopologyValidator validator = new TopologyValidator(config);
     List<String> results = validator.validate(topology);
@@ -63,19 +54,14 @@ public class TopologyValidationTest {
   }
 
   @Test
-  public void testInvalidExecutionWithFailedValidation() throws IOException {
+  public void testInvalidExecutionWithFailedValidation() {
 
     Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
 
-    Map<String, String> cliOps = new HashMap<>();
-    cliOps.put(BROKERS_OPTION, "");
-    cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
-
-    Properties props = new Properties();
-    props.put(
-        TOPOLOGY_VALIDATIONS_CONFIG,
-        Arrays.asList("topology.CamelCaseNameFormatValidation", "topic.PartitionNumberValidation"));
-    TopologyBuilderConfig config = new TopologyBuilderConfig(cliOps, props);
+    TopologyBuilderConfig config =
+        createTopologyBuilderConfig(
+                "com.purbon.kafka.topology.validation.topology.CamelCaseNameFormatValidation",
+                "com.purbon.kafka.topology.validation.topic.PartitionNumberValidation");
 
     TopologyValidator validator = new TopologyValidator(config);
     List<String> results = validator.validate(topology);
@@ -90,4 +76,40 @@ public class TopologyValidationTest {
     assertThat(results.get(4))
         .isEqualTo("Topic contextOrg.source.baz.topicF has an invalid number of partitions: 1");
   }
+
+  @Test
+  public void testUsingDeprecatedValidationsConfig() {
+
+    Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
+
+    TopologyBuilderConfig config =
+            createTopologyBuilderConfig(
+                    "topology.CamelCaseNameFormatValidation");
+
+    TopologyValidator validator = new TopologyValidator(config);
+    List<String> results = validator.validate(topology);
+    assertThat(results).hasSize(1);
+    assertThat(results.get(0)).isEqualTo("Project name does not follow the camelCase format: foo");
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void testUsingUnknwownClassName() {
+
+    Topology topology = parser.deserialise(TestUtils.getResourceFile("/descriptor.yaml"));
+
+    TopologyBuilderConfig config = createTopologyBuilderConfig("not.available.Validation");
+
+    TopologyValidator validator = new TopologyValidator(config);
+    validator.validate(topology);
+  }
+
+  private TopologyBuilderConfig createTopologyBuilderConfig(String... validations) {
+    Map<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+    cliOps.put(ADMIN_CLIENT_CONFIG_OPTION, "/fooBar");
+
+    Properties props = new Properties();
+    props.put(TOPOLOGY_VALIDATIONS_CONFIG, Arrays.asList(validations));
+    return new TopologyBuilderConfig(cliOps, props);
+  }
 }