Allow absolute path for truststore (#299)

* Added support for aboslute paths in the filesytem

* Updated documentation and added a test case for file system based loading

* Cleaned up a couple of issues pointed out by style check and static code analysis tools

* Apply suggestions from code review

---------

Co-authored-by: Ryan Skraba <ryan.skraba@aiven.io>

Author: madhavvishnubhatta

docs/sink-connector-config-options.rst

@@ -52,6 +52,15 @@ Connection
 
 ``http.ssl.truststore.location``
   Path to the SSL truststore file. Only JKS (Java KeyStore) format is supported.
+  
+  The truststore file can be located in the following ways (searched in order):
+  
+  1. **Class-based resource loading**: File bundled within the connector JAR
+  2. **Context classloader resource loading**: File accessible via the broader classpath (including Kafka Connect plugin directory)
+  3. **File system path**: 
+     
+     - **Absolute path**: Direct file system path (e.g., ``/path/to/truststore.jks``)
+     - **Relative path**: Path relative to the connector JAR's parent directory
 
   * Type: string
   * Default: null

src/main/java/io/aiven/kafka/connect/http/sender/TrustStoreLoader.java

@@ -81,16 +81,26 @@ private static InputStream tryFileSystem(final String path) {
             return null;
         }
         try {
-            final URL jarLocation = TrustStoreLoader.class.getProtectionDomain().getCodeSource().getLocation();
-            LOG.debug("JAR location: {}", jarLocation);
-            final Path jarPath = Paths.get(jarLocation.toURI());
-            final Path parentPath = jarPath.getParent();
-            if (parentPath == null) {
-                LOG.info("JAR has no parent directory, skipping file system lookup");
-                return null;
+            final Path truststorePath;
+            if (Paths.get(path).isAbsolute()) {
+                // Use absolute path directly
+                truststorePath = Paths.get(path);
+                LOG.info("Using absolute file system path: {}", truststorePath);
+            } else {
+                // Resolve relative path against JAR location
+                final URL jarLocation = TrustStoreLoader.class.getProtectionDomain().getCodeSource().getLocation();
+                LOG.info("JAR location: {}", jarLocation);
+                final Path jarPath = Paths.get(jarLocation.toURI());
+                LOG.info("Using relative path against the JAR location: {}", jarPath);
+                final Path parentPath = jarPath.getParent();
+                LOG.info("JAR parentPath: {}", parentPath);
+                if (parentPath == null) {
+                    LOG.info("JAR has no parent directory, skipping file system lookup");
+                    return null;
+                }
+                truststorePath = parentPath.resolve(path);
+                LOG.info("Trying relative file system path: {}", truststorePath);
             }
-            final Path truststorePath = parentPath.resolve(path.startsWith("/") ? path.substring(1) : path);
-            LOG.debug("Trying file system path: {}", truststorePath);
             final File truststoreFile = truststorePath.toFile();
             if (truststoreFile.exists()) {
                 LOG.info("Found via file system");

src/test/java/io/aiven/kafka/connect/http/sender/TrustStoreLoaderTest.java

@@ -16,9 +16,14 @@
 
 package io.aiven.kafka.connect.http.sender;
 
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
 import java.io.InputStream;
+import java.nio.file.Path;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -35,4 +40,19 @@ void findTrustStoreInputStreamReturnsNullForEmptyPath() {
         final InputStream result = TrustStoreLoader.findTrustStoreInputStream("");
         assertThat(result).isNull();
     }
+
+    @Test
+    void findTrustStoreInputStreamFindsFileWithAbsolutePath(@TempDir final Path tempDir) throws IOException {
+        // Create a temporary trust store file
+        final File trustStoreFile = tempDir.resolve("test-truststore.jks").toFile();
+        try (final FileOutputStream fos = new FileOutputStream(trustStoreFile)) {
+            fos.write("dummy truststore content".getBytes(java.nio.charset.StandardCharsets.UTF_8));
+        }
+
+        // Test with absolute path
+        final InputStream result = TrustStoreLoader.findTrustStoreInputStream(trustStoreFile.getAbsolutePath());
+        
+        assertThat(result).isNotNull();
+        result.close();
+    }
 }