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

Issue:
Schema Path is relative to the root directory path
Is there a reason why the schema path is a relative path to where topology file is present? 

Could we make it as an absolute path instead so that user can reference to a schema present even in a separate directory for organization purposes?

Pull request:
Add support for absolute paths when fetching schemas from disk
* **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.

closes #160  

Patch:
diff --git a/docs/futures/what-schema-management.rst b/docs/futures/what-schema-management.rst
index 5b6fefb..4966ba5 100644
--- a/docs/futures/what-schema-management.rst
+++ b/docs/futures/what-schema-management.rst
@@ -39,10 +39,36 @@ An example topology for managing schemas would look like this (*only the topics
 If using an example like this, for the topic _bar_ there is going to be an schema registered for key and value.
 Currently the schemas are managed as files and need to be accessible for the KTB tool, for example inside the git repository.
 
+A topology with absolute schema path would look like this:
+
+.. code-block:: YAML
+
+  ---
+    context: "context"
+    projects:
+      - name: "projectA"
+        topics:
+          - name: "foo"
+            schemas:
+              value.schema.file: "/kafka/schemas/foo-value.avsc"
+            config:
+              replication.factor: "1"
+              num.partitions: "1"
+
 *NOTE*: The path for the files is relative to the location of the of the topology.
 
 *NOTE*: Keep in mind that for a use case like in the example above you have to define the value for `schema.registry.url` in your properties file.
 
+Where should you put your schema files
+-----------
+
+We recommend keeping your schemas relative to the location of your topologies files as it is easier to map all things bundler near by than in
+distant locations.
+
+However, if preferred for organisational purposes the reader can specify any absolute path in the topology and the tool will prefer it as an schema location.
+
+*NOTE*: There is no way as of now to set a preferred location by configuration, so the user should write an absolute path where necessary in the topology file.
+
 Multiple schemas support per topic
 -----------
 
@@ -53,11 +79,6 @@ other options.
 
 The user should as well set the *record.type* for each of the keys as this is a value required when configuring the final subject.
 
-Supported Schema Registry
------------
-
-There are multiple options when using an Schema Registry with Kafka, currently only the Confluent Schema Registry is supported.
-
 Support multiple formats
 -----------
 
@@ -115,4 +136,9 @@ In the Schema management section is possible to set the schema compatibility lev
               replication.factor: "1"
               num.partitions: "1"
 
-**NOTE**: The compatibility level will be set before summit the registered schema file, this is done like this to easy transitions and migrations.
\ No newline at end of file
+**NOTE**: The compatibility level will be set before summit the registered schema file, this is done like this to easy transitions and migrations.
+
+Supported Schema Registry
+-----------
+
+There are multiple options when using an Schema Registry with Kafka, currently only the Confluent Schema Registry is supported.
\ No newline at end of file
diff --git a/src/main/java/com/purbon/kafka/topology/schemas/SchemaRegistryManager.java b/src/main/java/com/purbon/kafka/topology/schemas/SchemaRegistryManager.java
index e218433..b27bf0c 100644
--- a/src/main/java/com/purbon/kafka/topology/schemas/SchemaRegistryManager.java
+++ b/src/main/java/com/purbon/kafka/topology/schemas/SchemaRegistryManager.java
@@ -70,9 +70,11 @@ public class SchemaRegistryManager {
   }
 
   private Path schemaFilePath(String schemaFile) {
-    Path p = Paths.get(rootPath, schemaFile);
-    LOGGER.debug(p);
-    return p;
+    Path mayBeAbsolutePath = Paths.get(schemaFile);
+    Path path =
+        mayBeAbsolutePath.isAbsolute() ? mayBeAbsolutePath : Paths.get(rootPath, schemaFile);
+    LOGGER.debug(String.format("SchemaFile %s loaded from %s", schemaFile, path));
+    return path;
   }
 
   protected int save(String subjectName, String schemaType, String schemaString) {


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/schemas/SchemaRegistryManagerTest.java b/src/test/java/com/purbon/kafka/topology/schemas/SchemaRegistryManagerTest.java
index 4a62c68..283cdd2 100644
--- a/src/test/java/com/purbon/kafka/topology/schemas/SchemaRegistryManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/schemas/SchemaRegistryManagerTest.java
@@ -1,6 +1,8 @@
 package com.purbon.kafka.topology.schemas;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 import com.purbon.kafka.topology.schemas.SchemaRegistryManager.SchemaRegistryManagerException;
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
@@ -12,7 +14,12 @@ import java.net.URISyntaxException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
 
 public class SchemaRegistryManagerTest {
 
@@ -23,10 +30,16 @@ public class SchemaRegistryManagerTest {
   private SchemaRegistryClient client;
   private SchemaRegistryManager manager;
 
+  private Path rootDir;
+
+  @Rule public MockitoRule mockitoRule = MockitoJUnit.rule();
+
+  @Mock SchemaRegistryClient mockClient;
+
   @Before
   public void before() {
     client = new MockSchemaRegistryClient();
-    Path rootDir = Paths.get(System.getProperty("user.dir"), "target", "test-classes");
+    rootDir = Paths.get(System.getProperty("user.dir"), "target", "test-classes");
     manager = new SchemaRegistryManager(client, rootDir.toString());
   }
 
@@ -73,8 +86,19 @@ public class SchemaRegistryManagerTest {
   }
 
   @Test
-  public void shouldThrowAnExceptionWithValidRelativeFilePath() {
-    manager.register(subjectName, "schemas/bar-value.avsc", AvroSchema.TYPE);
+  public void shouldRegisterSchemasWithARelativePath() {
+    SchemaRegistryManager managerSpy = Mockito.spy(manager);
+    managerSpy.register(subjectName, "schemas/bar-value.avsc", AvroSchema.TYPE);
+    Path mayBeAbsolutePath = Paths.get(rootDir.toString(), "schemas/bar-value.avsc");
+    verify(managerSpy, times(1)).register(subjectName, mayBeAbsolutePath, AvroSchema.TYPE);
+  }
+
+  @Test
+  public void shouldRegisterSchemasWithAnAbsolutePath() {
+    SchemaRegistryManager managerSpy = Mockito.spy(manager);
+    Path mayBeAbsolutePath = Paths.get(rootDir.toString(), "schemas/bar-value.avsc");
+    managerSpy.register(subjectName, mayBeAbsolutePath.toString(), AvroSchema.TYPE);
+    verify(managerSpy, times(1)).register(subjectName, mayBeAbsolutePath, AvroSchema.TYPE);
   }
 
   @Test