SQOOP-1821: Sqoop2: External connector loading

(Veena Basavaraj via Abraham Elmahrek)

common/src/main/java/org/apache/sqoop/utils/ClassUtils.java

@@ -57,6 +57,8 @@ public static Class<?> loadClass(String className) {
           klass = ctxLoader.loadClass(className);
         } catch (ClassNotFoundException ex) {
           LOG.debug("Exception while load class: " + className, ex);
+          // wrapping it in runtime, to avoid chainging the signature of methods currently invoking this method
+          throw new RuntimeException(ex);
         }
       }
     }

common/src/test/java/org/apache/sqoop/utils/TestClassUtils.java

@@ -29,7 +29,7 @@
  */
 public class TestClassUtils {
 
-  @Test
+  @Test(expectedExceptions = Exception.class)
   public void testLoadClass() {
     assertNull(ClassUtils.loadClass("A"));
     assertEquals(A.class, ClassUtils.loadClass(A.class.getName()));

core/src/main/java/org/apache/sqoop/connector/ConnectorHandler.java

@@ -31,6 +31,7 @@
 import org.apache.sqoop.model.MFromConfig;
 import org.apache.sqoop.model.MLinkConfig;
 import org.apache.sqoop.model.MToConfig;
+import org.apache.sqoop.utils.ClassUtils;
 
 public final class ConnectorHandler {
 
@@ -71,8 +72,8 @@ public ConnectorHandler(URL configFileUrl) {
 
     Class<?> connectorClass = null;
     try {
-      connectorClass = Class.forName(connectorClassName);
-    } catch (ClassNotFoundException ex) {
+      connectorClass = ClassUtils.loadClass(connectorClassName);
+    } catch (Exception ex) {
       throw new SqoopException(ConnectorError.CONN_0005,
               connectorClassName, ex);
     }

core/src/main/java/org/apache/sqoop/connector/ConnectorManager.java

@@ -27,6 +27,7 @@
 import java.util.ResourceBundle;
 import java.util.Set;
 
+import org.apache.commons.lang.StringUtils;
 import org.apache.log4j.Logger;
 import org.apache.sqoop.common.SqoopException;
 import org.apache.sqoop.connector.spi.SqoopConnector;
@@ -158,6 +159,11 @@ public synchronized void initialize(boolean autoUpgrade) {
       LOG.trace("Begin connector manager initialization");
     }
 
+    // add external connectors into the class path
+    // NOTE: class loading happens later in the ConnectorHandler
+    ConnectorManagerUtils.addExternalConnectorsJarsToClasspath(SqoopConfiguration.getInstance().getContext()
+        .getString(ConfigurationConstants.EXTERNAL_CONNECTOR_LOAD_PATH, StringUtils.EMPTY));
+
     List<URL> connectorConfigs = ConnectorManagerUtils.getConnectorConfigs();
 
     LOG.info("Connector config urls: " + connectorConfigs);

core/src/main/java/org/apache/sqoop/connector/ConnectorManagerUtils.java

@@ -17,14 +17,22 @@
  */
 package org.apache.sqoop.connector;
 
+import org.apache.commons.lang.StringUtils;
 import org.apache.sqoop.common.SqoopException;
 import org.apache.sqoop.core.ConfigurationConstants;
 
+import java.io.File;
 import java.io.IOException;
+import java.net.MalformedURLException;
 import java.net.URL;
+import java.net.URLClassLoader;
 import java.util.ArrayList;
 import java.util.Enumeration;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
 
 /**
  * Utilities for ConnectorManager.
@@ -33,7 +41,7 @@ public class ConnectorManagerUtils {
 
   /**
    * Get a list of URLs of connectors that are installed.
-   * Check
+   *
    * @return List of URLs.
    */
   public static List<URL> getConnectorConfigs() {
@@ -41,9 +49,8 @@ public static List<URL> getConnectorConfigs() {
 
     try {
       // Check ConnectorManager classloader.
-      Enumeration<URL> appPathConfigs =
-          ConnectorManager.class.getClassLoader().getResources(
-              ConfigurationConstants.FILENAME_CONNECTOR_PROPERTIES);
+      Enumeration<URL> appPathConfigs = ConnectorManager.class.getClassLoader().getResources(
+          ConfigurationConstants.FILENAME_CONNECTOR_PROPERTIES);
       while (appPathConfigs.hasMoreElements()) {
         connectorConfigs.add(appPathConfigs.nextElement());
       }
@@ -51,8 +58,7 @@ public static List<URL> getConnectorConfigs() {
       // Check thread context classloader.
       ClassLoader ctxLoader = Thread.currentThread().getContextClassLoader();
       if (ctxLoader != null) {
-        Enumeration<URL> ctxPathConfigs = ctxLoader.getResources(
-            ConfigurationConstants.FILENAME_CONNECTOR_PROPERTIES);
+        Enumeration<URL> ctxPathConfigs = ctxLoader.getResources(ConfigurationConstants.FILENAME_CONNECTOR_PROPERTIES);
 
         while (ctxPathConfigs.hasMoreElements()) {
           URL configUrl = ctxPathConfigs.nextElement();
@@ -67,4 +73,73 @@ public static List<URL> getConnectorConfigs() {
 
     return connectorConfigs;
   }
+
+  public static Set<File> getConnectorJars(String path) {
+    if (StringUtils.isEmpty(path)) {
+      return null;
+    }
+    Set<File> jarFiles = new HashSet<File>();
+    File folder = new File(path);
+    if (folder.exists()) {
+      for (File file : folder.listFiles()) {
+        if (file.isDirectory()) {
+          jarFiles.addAll(getConnectorJars(file.getPath()));
+        }
+        if (file.getName().endsWith(".jar") && isConnectorJar(file)) {
+          jarFiles.add(file);
+        }
+      }
+    }
+    return jarFiles;
+  }
+
+  static boolean isConnectorJar(File file) {
+    try {
+      @SuppressWarnings("resource")
+      JarEntry entry = new JarFile(file).getJarEntry(ConfigurationConstants.FILENAME_CONNECTOR_PROPERTIES);
+      return entry != null;
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  public static void addExternalConnectorsJarsToClasspath(String path) {
+    if (StringUtils.isEmpty(path)) {
+      return;
+    }
+
+    ClassLoader currentThreadClassLoader = Thread.currentThread().getContextClassLoader();
+    if (currentThreadClassLoader != null) {
+
+      // Add the 'org.apache.sqoop.connector.external.loadpath' to the classpath
+      // Chain the current thread classloader
+      ExternalConnectorJarFileLoader connectorUrlClassLoader = new ExternalConnectorJarFileLoader(new URL[] {},
+          currentThreadClassLoader);
+      // the property always holds a path to the folder containing the jars
+      Set<File> connectorJars = getConnectorJars(path);
+      if (connectorJars != null && !connectorJars.isEmpty()) {
+        for (File jar : connectorJars) {
+          connectorUrlClassLoader.addJarFile(jar.getPath());
+        }
+        // Replace the thread classloader- assuming there is permission to do so
+        Thread.currentThread().setContextClassLoader(connectorUrlClassLoader);
+      }
+    }
+  }
+
+  public static class ExternalConnectorJarFileLoader extends URLClassLoader {
+    public ExternalConnectorJarFileLoader(URL[] urls, ClassLoader parent) {
+      super(urls, parent);
+    }
+
+    public void addJarFile(String path) {
+      String urlPath = "jar:file://" + path + "!/";
+      try {
+        addURL(new URL(urlPath));
+      } catch (MalformedURLException e) {
+        throw new RuntimeException(e);
+      }
+    }
+
+  }
 }

core/src/main/java/org/apache/sqoop/core/ConfigurationConstants.java

@@ -76,6 +76,13 @@ public final class ConfigurationConstants {
   public static final String DRIVER_AUTO_UPGRADE =
       "org.apache.sqoop.driver.autoupgrade";
 
+  /**
+   # Support loading external connector jars only
+   # The loader will look for sqoopconnector.properties file in the jar before loading
+   # "/path/to/external/connectors/": Add all the connector JARs in the specified folder
+   */
+  public static final String EXTERNAL_CONNECTOR_LOAD_PATH = "org.apache.sqoop.connector.external.loadpath";
+
   /**
    * Enable Sqoop App to kill Tomcat in case that it will fail to load.
    */

core/src/test/java/org/apache/sqoop/connector/TestConnectorManagerUtils.java

@@ -0,0 +1,64 @@
+package org.apache.sqoop.connector;
+
+import static org.testng.AssertJUnit.assertFalse;
+import static org.testng.AssertJUnit.assertTrue;
+import static org.testng.AssertJUnit.assertNull;
+import static org.testng.AssertJUnit.assertEquals;
+import org.apache.sqoop.utils.ClassUtils;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+import org.testng.AssertJUnit;
+
+import java.io.File;
+import java.net.URL;
+import java.util.List;
+import java.util.Set;
+
+public class TestConnectorManagerUtils {
+
+  private String workingDir;
+
+  @BeforeMethod(alwaysRun = true)
+  public void setUp() {
+    workingDir = System.getProperty("user.dir");
+  }
+
+  @Test
+  public void testGetConnectorJarsNullPath() {
+    Set<File> files = ConnectorManagerUtils.getConnectorJars(null);
+    assertNull(files);
+  }
+
+  @Test
+  public void testGetConnectorJarsNonNullPath() {
+    String path = workingDir + "/src/test/resources";
+    Set<File> files = ConnectorManagerUtils.getConnectorJars(path);
+    assertEquals(1, files.size());
+  }
+
+  @Test
+  public void testIsConnectorJar() {
+    String path = workingDir + "/src/test/resources/test-connector.jar";
+    File connectorJar = new File(path);
+    assertTrue(connectorJar.exists());
+    assertTrue(ConnectorManagerUtils.isConnectorJar(connectorJar));
+  }
+
+  @Test
+  public void testIsNotConnectorJar() {
+    String path = workingDir + "/src/test/resources/test-non-connector.jar";
+    File file = new File(path);
+    assertTrue(file.exists());
+    assertFalse(ConnectorManagerUtils.isConnectorJar(file));
+  }
+
+  @Test
+  public void testAddExternalConnectorJarToClasspath() {
+    String path = workingDir + "/src/test/resources";
+    ConnectorManagerUtils.addExternalConnectorsJarsToClasspath(path);
+    List<URL> urls = ConnectorManagerUtils.getConnectorConfigs();
+    assertEquals(1, urls.size());
+    ClassUtils.loadClass("org.apache.sqoop.connector.jdbc.GenericJdbcConnector");
+  }
+
+}

dist/src/main/server/conf/sqoop.properties

@@ -163,4 +163,10 @@ org.apache.sqoop.execution.engine=org.apache.sqoop.execution.mapreduce.Mapreduce
 #
 #org.apache.sqoop.security.authorization.handler=org.apache.sqoop.security.Authorization.DefaultAuthorizationHandler
 #org.apache.sqoop.security.authorization.access_controller=org.apache.sqoop.security.Authorization.DefaultAuthorizationAccessController
-#org.apache.sqoop.security.authorization.validator=org.apache.sqoop.security.Authorization.DefaultAuthorizationValidator
+#org.apache.sqoop.security.authorization.validator=org.apache.sqoop.security.Authorization.DefaultAuthorizationValidator
+
+
+# External connectors load path
+# "/path/to/external/connectors/": Add all the connector JARs in the specified folder
+#
+org.apache.sqoop.connector.external.loadpath=

test/src/main/java/org/apache/sqoop/test/minicluster/SqoopMiniCluster.java

@@ -132,6 +132,7 @@ protected void prepareTemporaryPath() throws IOException {
     mapToProperties(sqoopProperties, getSecurityConfiguration());
     mapToProperties(sqoopProperties, getConnectorManagerConfiguration());
     mapToProperties(sqoopProperties, getDriverManagerConfiguration());
+    mapToProperties(sqoopProperties, getExternalConnectorLoadPathConfiguration());
 
     FileUtils.writeLines(f, sqoopProperties);
 
@@ -212,17 +213,19 @@ protected Map<String, String> getSecurityConfiguration() {
 
   protected Map<String, String> getConnectorManagerConfiguration() {
     Map<String, String> properties = new HashMap<String, String>();
-
-    properties.put("org.apache.sqoop.connector.autoupgrade", "true");
-
+    properties.put(ConfigurationConstants.CONNECTOR_AUTO_UPGRADE, "true");
     return properties;
   }
 
   protected Map<String, String> getDriverManagerConfiguration() {
     Map<String, String> properties = new HashMap<String, String>();
+    properties.put(ConfigurationConstants.DRIVER_AUTO_UPGRADE, "true");
+    return properties;
+  }
 
-    properties.put("org.apache.sqoop.driver.autoupgrade", "true");
-
+  protected Map<String, String> getExternalConnectorLoadPathConfiguration() {
+    Map<String, String> properties = new HashMap<String, String>();
+    properties.put(ConfigurationConstants.EXTERNAL_CONNECTOR_LOAD_PATH, "");
     return properties;
   }
 }