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:
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);
+  }
 }