custom-dynclass-loader (#216)

* custom-dynclass-loader

- we produce a `-hive` and a not hive version of the connector.  Due to how
we strip the packages, some Kafka Connect clusters can throw a NoClassDefFound
error when it cannot find hadoop/hive (e.g. when using a rest catalog). This is fine,
however the refleciton code we use only covers ClassPathNotFound to nulls.  This fixes 
the non-hive version from throwing in some clusters.

kafka-connect/src/main/java/io/tabular/iceberg/connect/data/Utilities.java

@@ -72,11 +72,9 @@ public static Catalog loadCatalog(IcebergSinkConfig config) {
 
   // use reflection here to avoid requiring Hadoop as a dependency
   private static Object loadHadoopConfig(IcebergSinkConfig config) {
-    Class<?> configClass =
-        DynClasses.builder().impl("org.apache.hadoop.hdfs.HdfsConfiguration").orNull().build();
+    Class<?> configClass = dynamicallyLoad("org.apache.hadoop.hdfs.HdfsConfiguration");
     if (configClass == null) {
-      configClass =
-          DynClasses.builder().impl("org.apache.hadoop.conf.Configuration").orNull().build();
+      configClass = dynamicallyLoad("org.apache.hadoop.conf.Configuration");
     }
 
     if (configClass == null) {
@@ -250,5 +248,21 @@ public static TaskWriter<Record> createTableWriter(
     return writer;
   }
 
+  /**
+   * Dynamically load hive/hadoop configs to avoid packaging them with the distribution. Gradle
+   * strips hadoop from the classpath which will cause a NoClassDefFoundError to be thrown when
+   * using the version without Hive, so intercept that exception to maintain the underlying
+   * DynClass.builder.impl(...).orNull() behavior.
+   */
+  private static Class<?> dynamicallyLoad(String className) {
+    Class<?> configClass;
+    try {
+      configClass = DynClasses.builder().impl(className).orNull().build();
+    } catch (NoClassDefFoundError e) {
+      configClass = null;
+    }
+    return configClass;
+  }
+
   private Utilities() {}
 }