SQOOP-3309: Implement HiveServer2 client
(Szabolcs Vasas via Boglarka Egyed)
diff --git a/build.xml b/build.xml
index 7f68b57..a85705f 100644
--- a/build.xml
+++ b/build.xml
@@ -734,6 +734,9 @@
<!-- enable asserts in tests -->
<jvmarg value="-ea" />
+ <!-- We need to disable asserts in HadoopThriftAuthBridge to be able to run HiveMiniCluster tests. -->
+ <jvmarg value="-da:org.apache.hadoop.hive.thrift.HadoopThriftAuthBridge" />
+
<jvmarg value="${remoteDebugJvmArgs}"/>
<sysproperty key="test.build.data" value="${build.test}/data"/>
diff --git a/ivy.xml b/ivy.xml
index 6be4fa2..6af94d9 100644
--- a/ivy.xml
+++ b/ivy.xml
@@ -195,6 +195,8 @@
<exclude org="org.apache.avro" module="avro" />
</dependency>
+ <dependency org="org.apache.hive" name="hive-jdbc" rev="${hcatalog.version}" conf="common->default" />
+
<dependency org="org.apache.hive.hcatalog" name="hive-hcatalog-core"
rev="${hcatalog.version}" conf="common->default">
<artifact name="hive-hcatalog-core" type="jar"/>
diff --git a/src/docs/user/hive-args.txt b/src/docs/user/hive-args.txt
index 441f54e..7509564 100644
--- a/src/docs/user/hive-args.txt
+++ b/src/docs/user/hive-args.txt
@@ -43,5 +43,10 @@
Hive type for configured columns. If specify commas in\
this argument, use URL encoded keys and values, for example,\
use DECIMAL(1%2C%201) instead of DECIMAL(1, 1).
++\--hs2-url+ The JDBC connection string to HiveServer2 as you would specify in Beeline. If you use this option with \
+ --hive-import then Sqoop will try to connect to HiveServer2 instead of using Hive CLI.
++\--hs2-user+ The user for creating the JDBC connection to HiveServer2. The default is the current OS user.
++\--hs2-keytab+ The path to the keytab file of the user connecting to HiveServer2. If you choose another \
+ HiveServer2 user (with --hs2-user) then --hs2-keytab has to be also specified otherwise it can be omitted.
--------------------------------------------------------------------------
diff --git a/src/docs/user/hive.txt b/src/docs/user/hive.txt
index 3dc8bb4..f8f7c27 100644
--- a/src/docs/user/hive.txt
+++ b/src/docs/user/hive.txt
@@ -35,12 +35,22 @@
operation defining your columns using Hive's types, and a +LOAD DATA INPATH+
statement to move the data files into Hive's warehouse directory.
-The script will be executed by calling
+The script can be executed in two ways:
+
+- By the default the script will be executed by calling
the installed copy of hive on the machine where Sqoop is run. If you have
multiple Hive installations, or +hive+ is not in your +$PATH+, use the
*+\--hive-home+* option to identify the Hive installation directory.
Sqoop will use +$HIVE_HOME/bin/hive+ from here.
+- If the user specifies the *+\--hs2-url+* parameter then the script will
+ be sent to HiveServer2 through a JDBC connection. Note that the data itself
+ will not be transferred via the JDBC connection it is written directly to HDFS
+ just like in case of the default hive import. As HiveServer2 provides proper
+ authorization and auditing features it is recommended to use this instead of
+ the default. Currently only Kerberos authentication and text file format is
+ supported with this option.
+
NOTE: This function is incompatible with +\--as-avrodatafile+ and
+\--as-sequencefile+.
diff --git a/src/java/org/apache/sqoop/SqoopOptions.java b/src/java/org/apache/sqoop/SqoopOptions.java
index 651cebd..d9984af 100644
--- a/src/java/org/apache/sqoop/SqoopOptions.java
+++ b/src/java/org/apache/sqoop/SqoopOptions.java
@@ -449,6 +449,15 @@
private String metaUsername;
private String metaPassword;
+ @StoredAsProperty("hs2.url")
+ private String hs2Url;
+
+ @StoredAsProperty("hs2.user")
+ private String hs2User;
+
+ @StoredAsProperty("hs2.keytab")
+ private String hs2Keytab;
+
public SqoopOptions() {
initDefaults(null);
}
@@ -2892,5 +2901,29 @@
this.metaPassword = metaPassword;
}
+ public String getHs2Url() {
+ return hs2Url;
+ }
+
+ public void setHs2Url(String hs2Url) {
+ this.hs2Url = hs2Url;
+ }
+
+ public String getHs2User() {
+ return hs2User;
+ }
+
+ public void setHs2User(String hs2User) {
+ this.hs2User = hs2User;
+ }
+
+ public String getHs2Keytab() {
+ return hs2Keytab;
+ }
+
+ public void setHs2Keytab(String hs2Keytab) {
+ this.hs2Keytab = hs2Keytab;
+ }
+
}
diff --git a/src/java/org/apache/sqoop/hive/HiveClient.java b/src/java/org/apache/sqoop/hive/HiveClient.java
new file mode 100644
index 0000000..994f858
--- /dev/null
+++ b/src/java/org/apache/sqoop/hive/HiveClient.java
@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import java.io.IOException;
+
+public interface HiveClient {
+
+ void importTable() throws IOException;
+
+ void createTable() throws IOException;
+}
diff --git a/src/java/org/apache/sqoop/hive/HiveClientCommon.java b/src/java/org/apache/sqoop/hive/HiveClientCommon.java
new file mode 100644
index 0000000..952db6d
--- /dev/null
+++ b/src/java/org/apache/sqoop/hive/HiveClientCommon.java
@@ -0,0 +1,107 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.FileOutputCommitter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.io.CodecMap;
+
+import java.io.IOException;
+
+/**
+ * Class containing the common logic for different HiveClient implementations.
+ */
+public class HiveClientCommon {
+
+ public static final Log LOG = LogFactory.getLog(HiveClientCommon.class.getName());
+
+ /**
+ * If we used a MapReduce-based upload of the data, remove the _logs dir
+ * from where we put it, before running Hive LOAD DATA INPATH.
+ */
+ public void removeTempLogs(Configuration configuration, Path tablePath) throws IOException {
+ FileSystem fs = tablePath.getFileSystem(configuration);
+ Path logsPath = new Path(tablePath, "_logs");
+ if (fs.exists(logsPath)) {
+ LOG.info("Removing temporary files from import process: " + logsPath);
+ if (!fs.delete(logsPath, true)) {
+ LOG.warn("Could not delete temporary files; "
+ + "continuing with import, but it may fail.");
+ }
+ }
+ }
+
+ /**
+ * Clean up after successful HIVE import.
+ *
+ * @param outputPath path to the output directory
+ * @throws IOException
+ */
+ public void cleanUp(Configuration configuration, Path outputPath) throws IOException {
+ FileSystem fs = outputPath.getFileSystem(configuration);
+
+ // HIVE is not always removing input directory after LOAD DATA statement
+ // (which is our export directory). We're removing export directory in case
+ // that is blank for case that user wants to periodically populate HIVE
+ // table (for example with --hive-overwrite).
+ try {
+ if (outputPath != null && fs.exists(outputPath)) {
+ FileStatus[] statuses = fs.listStatus(outputPath);
+ if (statuses.length == 0) {
+ LOG.info("Export directory is empty, removing it.");
+ fs.delete(outputPath, true);
+ } else if (statuses.length == 1 && statuses[0].getPath().getName().equals(FileOutputCommitter.SUCCEEDED_FILE_NAME)) {
+ LOG.info("Export directory is contains the _SUCCESS file only, removing the directory.");
+ fs.delete(outputPath, true);
+ } else {
+ LOG.info("Export directory is not empty, keeping it.");
+ }
+ }
+ } catch(IOException e) {
+ LOG.error("Issue with cleaning (safe to ignore)", e);
+ }
+ }
+
+ public void indexLzoFiles(SqoopOptions sqoopOptions, Path finalPath) throws IOException {
+ String codec = sqoopOptions.getCompressionCodec();
+ if (codec != null && (codec.equals(CodecMap.LZOP)
+ || codec.equals(CodecMap.getCodecClassName(CodecMap.LZOP)))) {
+ try {
+ Tool tool = ReflectionUtils.newInstance(Class.
+ forName("com.hadoop.compression.lzo.DistributedLzoIndexer").
+ asSubclass(Tool.class), sqoopOptions.getConf());
+ ToolRunner.run(sqoopOptions.getConf(), tool,
+ new String[]{finalPath.toString()});
+ } catch (Exception ex) {
+ LOG.error("Error indexing lzo files", ex);
+ throw new IOException("Error indexing lzo files", ex);
+ }
+ }
+ }
+
+}
diff --git a/src/java/org/apache/sqoop/hive/HiveClientFactory.java b/src/java/org/apache/sqoop/hive/HiveClientFactory.java
new file mode 100644
index 0000000..67de8f0
--- /dev/null
+++ b/src/java/org/apache/sqoop/hive/HiveClientFactory.java
@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.authentication.KerberosAuthenticator;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+import org.apache.sqoop.db.decorator.KerberizedConnectionFactoryDecorator;
+import org.apache.sqoop.manager.ConnManager;
+
+import java.io.IOException;
+
+import static org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.commons.lang3.StringUtils.isEmpty;
+
+public class HiveClientFactory {
+
+ private final HiveServer2ConnectionFactoryInitializer connectionFactoryInitializer;
+
+ public HiveClientFactory(HiveServer2ConnectionFactoryInitializer connectionFactoryInitializer) {
+ this.connectionFactoryInitializer = connectionFactoryInitializer;
+ }
+
+ public HiveClientFactory() {
+ this(new HiveServer2ConnectionFactoryInitializer());
+ }
+
+ public HiveClient createHiveClient(SqoopOptions sqoopOptions, ConnManager connManager) {
+ if (useHiveCli(sqoopOptions)) {
+ return createHiveImport(sqoopOptions, connManager);
+ } else {
+ return createHiveServer2Client(sqoopOptions, connManager);
+ }
+ }
+
+ private HiveClient createHiveImport(SqoopOptions sqoopOptions, ConnManager connManager) {
+ return new HiveImport(sqoopOptions, connManager, sqoopOptions.getConf(), false);
+ }
+
+ private HiveClient createHiveServer2Client(SqoopOptions sqoopOptions, ConnManager connManager) {
+ TableDefWriter tableDefWriter = createTableDefWriter(sqoopOptions, connManager);
+ JdbcConnectionFactory hs2JdbcConnectionFactory = connectionFactoryInitializer.createJdbcConnectionFactory(sqoopOptions);
+ return new HiveServer2Client(sqoopOptions, tableDefWriter, hs2JdbcConnectionFactory);
+ }
+
+ TableDefWriter createTableDefWriter(SqoopOptions sqoopOptions, ConnManager connManager) {
+ return new TableDefWriter(sqoopOptions, connManager, sqoopOptions.getTableName(), sqoopOptions.getHiveTableName(), sqoopOptions.getConf(), false);
+ }
+
+ private boolean useHiveCli(SqoopOptions sqoopOptions) {
+ return StringUtils.isEmpty(sqoopOptions.getHs2Url());
+ }
+
+}
diff --git a/src/java/org/apache/sqoop/hive/HiveImport.java b/src/java/org/apache/sqoop/hive/HiveImport.java
index c272911..5da00a7 100644
--- a/src/java/org/apache/sqoop/hive/HiveImport.java
+++ b/src/java/org/apache/sqoop/hive/HiveImport.java
@@ -31,16 +31,9 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapred.FileOutputCommitter;
-import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.Shell;
-import org.apache.hadoop.util.ToolRunner;
-import org.apache.hadoop.util.Tool;
-import org.apache.sqoop.io.CodecMap;
import org.apache.sqoop.util.Executor;
import org.apache.sqoop.util.LoggingAsyncSink;
import org.apache.sqoop.util.SubprocessSecurityManager;
@@ -54,7 +47,7 @@
* to Hive itself as well as orchestrating the use of the other classes in this
* package.
*/
-public class HiveImport {
+public class HiveImport implements HiveClient {
public static final Log LOG = LogFactory.getLog(HiveImport.class.getName());
@@ -62,6 +55,7 @@
private ConnManager connManager;
private Configuration configuration;
private boolean generateOnly;
+ private HiveClientCommon hiveClientCommon;
private static boolean testMode = false;
public static boolean getTestMode() {
@@ -77,13 +71,17 @@
"org.apache.hadoop.hive.cli.CliDriver";
public HiveImport(final SqoopOptions opts, final ConnManager connMgr,
- final Configuration conf, final boolean generateOnly) {
+ final Configuration conf, final boolean generateOnly, final HiveClientCommon hiveClientCommon) {
this.options = opts;
this.connManager = connMgr;
this.configuration = conf;
this.generateOnly = generateOnly;
+ this.hiveClientCommon = hiveClientCommon;
}
+ public HiveImport(SqoopOptions opts, ConnManager connMgr, Configuration conf, boolean generateOnly) {
+ this(opts, connMgr, conf, generateOnly, new HiveClientCommon());
+ }
/**
* @return the filename of the hive executable to run to do the import
@@ -111,27 +109,11 @@
}
/**
- * If we used a MapReduce-based upload of the data, remove the _logs dir
- * from where we put it, before running Hive LOAD DATA INPATH.
- */
- private void removeTempLogs(Path tablePath) throws IOException {
- FileSystem fs = tablePath.getFileSystem(configuration);
- Path logsPath = new Path(tablePath, "_logs");
- if (fs.exists(logsPath)) {
- LOG.info("Removing temporary files from import process: " + logsPath);
- if (!fs.delete(logsPath, true)) {
- LOG.warn("Could not delete temporary files; "
- + "continuing with import, but it may fail.");
- }
- }
- }
-
- /**
* @return true if we're just generating the DDL for the import, but
* not actually running it (i.e., --generate-only mode). If so, don't
* do any side-effecting actions in Hive.
*/
- private boolean isGenerateOnly() {
+ boolean isGenerateOnly() {
return generateOnly;
}
@@ -181,8 +163,6 @@
}
// generate the HQL statements to run.
- // reset the connection as it might have timed out
- connManager.discardConnection(true);
TableDefWriter tableWriter = new TableDefWriter(options, connManager,
inputTableName, outputTableName,
configuration, !debugMode);
@@ -191,23 +171,10 @@
Path finalPath = tableWriter.getFinalPath();
if (!isGenerateOnly()) {
- removeTempLogs(finalPath);
+ hiveClientCommon.removeTempLogs(configuration, finalPath);
LOG.info("Loading uploaded data into Hive");
- String codec = options.getCompressionCodec();
- if (codec != null && (codec.equals(CodecMap.LZOP)
- || codec.equals(CodecMap.getCodecClassName(CodecMap.LZOP)))) {
- try {
- Tool tool = ReflectionUtils.newInstance(Class.
- forName("com.hadoop.compression.lzo.DistributedLzoIndexer").
- asSubclass(Tool.class), configuration);
- ToolRunner.run(configuration, tool,
- new String[] { finalPath.toString() });
- } catch (Exception ex) {
- LOG.error("Error indexing lzo files", ex);
- throw new IOException("Error indexing lzo files", ex);
- }
- }
+ hiveClientCommon.indexLzoFiles(options, finalPath);
}
// write them to a script file.
@@ -242,7 +209,7 @@
LOG.info("Hive import complete.");
- cleanUp(finalPath);
+ hiveClientCommon.cleanUp(configuration, finalPath);
}
} finally {
if (!isGenerateOnly()) {
@@ -256,37 +223,6 @@
}
}
- /**
- * Clean up after successful HIVE import.
- *
- * @param outputPath path to the output directory
- * @throws IOException
- */
- private void cleanUp(Path outputPath) throws IOException {
- FileSystem fs = outputPath.getFileSystem(configuration);
-
- // HIVE is not always removing input directory after LOAD DATA statement
- // (which is our export directory). We're removing export directory in case
- // that is blank for case that user wants to periodically populate HIVE
- // table (for example with --hive-overwrite).
- try {
- if (outputPath != null && fs.exists(outputPath)) {
- FileStatus[] statuses = fs.listStatus(outputPath);
- if (statuses.length == 0) {
- LOG.info("Export directory is empty, removing it.");
- fs.delete(outputPath, true);
- } else if (statuses.length == 1 && statuses[0].getPath().getName().equals(FileOutputCommitter.SUCCEEDED_FILE_NAME)) {
- LOG.info("Export directory is contains the _SUCCESS file only, removing the directory.");
- fs.delete(outputPath, true);
- } else {
- LOG.info("Export directory is not empty, keeping it.");
- }
- }
- } catch(IOException e) {
- LOG.error("Issue with cleaning (safe to ignore)", e);
- }
- }
-
@SuppressWarnings("unchecked")
/**
* Execute the script file via Hive.
@@ -398,5 +334,28 @@
return newArgs.toArray(new String[newArgs.size()]);
}
+
+ @Override
+ public void importTable() throws IOException {
+ importTable(options.getTableName(), options.getHiveTableName(), false);
+ }
+
+ @Override
+ public void createTable() throws IOException {
+ importTable(options.getTableName(), options.getHiveTableName(), true);
+ }
+
+ SqoopOptions getOptions() {
+ return options;
+ }
+
+ ConnManager getConnManager() {
+ return connManager;
+ }
+
+ Configuration getConfiguration() {
+ return configuration;
+ }
+
}
diff --git a/src/java/org/apache/sqoop/hive/HiveServer2Client.java b/src/java/org/apache/sqoop/hive/HiveServer2Client.java
new file mode 100644
index 0000000..c4976c1
--- /dev/null
+++ b/src/java/org/apache/sqoop/hive/HiveServer2Client.java
@@ -0,0 +1,113 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.Path;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+
+import static java.util.Arrays.asList;
+
+public class HiveServer2Client implements HiveClient {
+
+ private static final Log LOG = LogFactory.getLog(HiveServer2Client.class.getName());
+
+ private final SqoopOptions sqoopOptions;
+
+ private final TableDefWriter tableDefWriter;
+
+ private final JdbcConnectionFactory hs2ConnectionFactory;
+
+ private final HiveClientCommon hiveClientCommon;
+
+ public HiveServer2Client(SqoopOptions sqoopOptions, TableDefWriter tableDefWriter, JdbcConnectionFactory hs2ConnectionFactory, HiveClientCommon hiveClientCommon) {
+ this.sqoopOptions = sqoopOptions;
+ this.tableDefWriter = tableDefWriter;
+ this.hs2ConnectionFactory = hs2ConnectionFactory;
+ this.hiveClientCommon = hiveClientCommon;
+ }
+
+ public HiveServer2Client(SqoopOptions sqoopOptions, TableDefWriter tableDefWriter, JdbcConnectionFactory hs2ConnectionFactory) {
+ this(sqoopOptions, tableDefWriter, hs2ConnectionFactory, new HiveClientCommon());
+ }
+
+ @Override
+ public void importTable() throws IOException {
+ LOG.info("Loading uploaded data into Hive.");
+ String createTableStmt = tableDefWriter.getCreateTableStmt();
+ String loadDataStmt = tableDefWriter.getLoadDataStmt();
+ executeHiveImport(asList(createTableStmt, loadDataStmt));
+ LOG.info("Hive import complete.");
+ }
+
+ @Override
+ public void createTable() throws IOException {
+ LOG.info("Creating Hive table: " + tableDefWriter.getOutputTableName());
+ String createTableStmt = tableDefWriter.getCreateTableStmt();
+ executeHiveImport(asList(createTableStmt));
+ LOG.info("Hive table is successfully created.");
+ }
+
+ void executeHiveImport(List<String> commands) throws IOException {
+ Path finalPath = tableDefWriter.getFinalPath();
+
+ hiveClientCommon.removeTempLogs(sqoopOptions.getConf(), finalPath);
+
+ hiveClientCommon.indexLzoFiles(sqoopOptions, finalPath);
+
+ try {
+ executeCommands(commands);
+ } catch (SQLException e) {
+ throw new RuntimeException("Error executing Hive import.", e);
+ }
+
+ hiveClientCommon.cleanUp(sqoopOptions.getConf(), finalPath);
+ }
+
+ void executeCommands(List<String> commands) throws SQLException {
+ try (Connection hs2Connection = hs2ConnectionFactory.createConnection()) {
+ for (String command : commands) {
+ LOG.debug("Executing command: " + command);
+ try (PreparedStatement statement = hs2Connection.prepareStatement(command)) {
+ statement.execute();
+ }
+ }
+ }
+ }
+
+ SqoopOptions getSqoopOptions() {
+ return sqoopOptions;
+ }
+
+ TableDefWriter getTableDefWriter() {
+ return tableDefWriter;
+ }
+
+ JdbcConnectionFactory getHs2ConnectionFactory() {
+ return hs2ConnectionFactory;
+ }
+}
diff --git a/src/java/org/apache/sqoop/hive/HiveServer2ConnectionFactory.java b/src/java/org/apache/sqoop/hive/HiveServer2ConnectionFactory.java
new file mode 100644
index 0000000..10515a0
--- /dev/null
+++ b/src/java/org/apache/sqoop/hive/HiveServer2ConnectionFactory.java
@@ -0,0 +1,60 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.sqoop.db.DriverManagerJdbcConnectionFactory;
+
+import java.io.IOException;
+import java.sql.Connection;
+
+import static org.apache.commons.lang3.StringUtils.EMPTY;
+
+public class HiveServer2ConnectionFactory extends DriverManagerJdbcConnectionFactory {
+
+ private static final Log LOG = LogFactory.getLog(HiveServer2ConnectionFactory.class.getName());
+
+ private static final String HS2_DRIVER_CLASS = "org.apache.hive.jdbc.HiveDriver";
+
+ public HiveServer2ConnectionFactory(String connectionString, String username, String password) {
+ super(HS2_DRIVER_CLASS, connectionString, username, password);
+ }
+
+ public HiveServer2ConnectionFactory(String connectionString, String username) {
+ this(connectionString, username, null);
+ }
+
+ @Override
+ public Connection createConnection() {
+ LOG.info("Creating connection to HiveServer2 as: " + getCurrentUser());
+ return super.createConnection();
+ }
+
+ private String getCurrentUser() {
+ try {
+ return UserGroupInformation.getCurrentUser().toString();
+ } catch (IOException e) {
+ LOG.error("Unable to determine current user.", e);
+ }
+ return EMPTY;
+ }
+
+}
diff --git a/src/java/org/apache/sqoop/hive/HiveServer2ConnectionFactoryInitializer.java b/src/java/org/apache/sqoop/hive/HiveServer2ConnectionFactoryInitializer.java
new file mode 100644
index 0000000..1d959f9
--- /dev/null
+++ b/src/java/org/apache/sqoop/hive/HiveServer2ConnectionFactoryInitializer.java
@@ -0,0 +1,64 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.authentication.KerberosAuthenticator;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+import org.apache.sqoop.db.decorator.KerberizedConnectionFactoryDecorator;
+
+import java.io.IOException;
+
+import static org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.commons.lang3.StringUtils.isEmpty;
+
+public class HiveServer2ConnectionFactoryInitializer {
+
+ public JdbcConnectionFactory createJdbcConnectionFactory(SqoopOptions sqoopOptions) {
+ String connectionUsername = determineConnectionUsername(sqoopOptions);
+ JdbcConnectionFactory connectionFactory = new HiveServer2ConnectionFactory(sqoopOptions.getHs2Url(), connectionUsername);
+ if (useKerberizedConnection(sqoopOptions)) {
+ KerberosAuthenticator authenticator = createKerberosAuthenticator(sqoopOptions);
+ connectionFactory = new KerberizedConnectionFactoryDecorator(connectionFactory, authenticator);
+ }
+ return connectionFactory;
+ }
+
+ private String determineConnectionUsername(SqoopOptions sqoopOptions) {
+ if (!isEmpty(sqoopOptions.getHs2User())) {
+ return sqoopOptions.getHs2User();
+ }
+ try {
+ return UserGroupInformation.getLoginUser().getUserName();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to determine login user.", e);
+ }
+ }
+
+ private KerberosAuthenticator createKerberosAuthenticator(SqoopOptions sqoopOptions) {
+ return new KerberosAuthenticator(sqoopOptions.getConf(), sqoopOptions.getHs2User(), sqoopOptions.getHs2Keytab());
+ }
+
+ private boolean useKerberizedConnection(SqoopOptions sqoopOptions) {
+ return !isBlank(sqoopOptions.getHs2Keytab());
+ }
+
+}
diff --git a/src/java/org/apache/sqoop/hive/TableDefWriter.java b/src/java/org/apache/sqoop/hive/TableDefWriter.java
index b7a25b7..27d988c 100644
--- a/src/java/org/apache/sqoop/hive/TableDefWriter.java
+++ b/src/java/org/apache/sqoop/hive/TableDefWriter.java
@@ -96,6 +96,7 @@
* @return the CREATE TABLE statement for the table to load into hive.
*/
public String getCreateTableStmt() throws IOException {
+ resetConnManager();
Map<String, Integer> columnTypes;
Properties userMapping = options.getMapColumnHive();
Boolean isHiveExternalTableSet = !StringUtils.isBlank(options.getHiveExternalTableDir());
@@ -286,5 +287,38 @@
return String.format("\\%03o", charNum);
}
+ /**
+ * The JDBC connection owned by the ConnManager has been most probably opened when the import was started
+ * so it might have timed out by the time TableDefWriter methods are invoked which happens at the end of import.
+ * The task of this method is to discard the current connection held by ConnManager to make sure
+ * that TableDefWriter will have a working one.
+ */
+ private void resetConnManager() {
+ this.connManager.discardConnection(true);
+ }
+
+ SqoopOptions getOptions() {
+ return options;
+ }
+
+ ConnManager getConnManager() {
+ return connManager;
+ }
+
+ Configuration getConfiguration() {
+ return configuration;
+ }
+
+ String getInputTableName() {
+ return inputTableName;
+ }
+
+ String getOutputTableName() {
+ return outputTableName;
+ }
+
+ boolean isCommentsEnabled() {
+ return commentsEnabled;
+ }
}
diff --git a/src/java/org/apache/sqoop/tool/BaseSqoopTool.java b/src/java/org/apache/sqoop/tool/BaseSqoopTool.java
index b02e4fe..783651a 100644
--- a/src/java/org/apache/sqoop/tool/BaseSqoopTool.java
+++ b/src/java/org/apache/sqoop/tool/BaseSqoopTool.java
@@ -18,6 +18,8 @@
package org.apache.sqoop.tool;
+import static java.lang.String.format;
+
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
@@ -131,6 +133,9 @@
public static final String HCATALOG_STORAGE_STANZA_ARG =
"hcatalog-storage-stanza";
public static final String HCATALOG_HOME_ARG = "hcatalog-home";
+ public static final String HS2_URL_ARG = "hs2-url";
+ public static final String HS2_USER_ARG = "hs2-user";
+ public static final String HS2_KEYTAB_ARG = "hs2-keytab";
public static final String MAPREDUCE_JOB_NAME = "mapreduce-job-name";
public static final String NUM_MAPPERS_ARG = "num-mappers";
public static final String NUM_MAPPERS_SHORT_ARG = "m";
@@ -609,6 +614,21 @@
+ " types.")
.withLongOpt(MAP_COLUMN_HIVE)
.create());
+ hiveOpts.addOption(OptionBuilder
+ .hasArg()
+ .withDescription("The URL to the HiveServer2.")
+ .withLongOpt(HS2_URL_ARG)
+ .create());
+ hiveOpts.addOption(OptionBuilder
+ .hasArg()
+ .withDescription("The user/principal for HiveServer2.")
+ .withLongOpt(HS2_USER_ARG)
+ .create());
+ hiveOpts.addOption(OptionBuilder
+ .hasArg()
+ .withDescription("The location of the keytab of the HiveServer2 user.")
+ .withLongOpt(HS2_KEYTAB_ARG)
+ .create());
return hiveOpts;
}
@@ -1238,6 +1258,15 @@
if (in.hasOption(HIVE_EXTERNAL_TABLE_LOCATION_ARG)) {
out.setHiveExternalTableDir(in.getOptionValue(HIVE_EXTERNAL_TABLE_LOCATION_ARG));
}
+ if (in.hasOption(HS2_URL_ARG)) {
+ out.setHs2Url(in.getOptionValue(HS2_URL_ARG));
+ }
+ if (in.hasOption(HS2_USER_ARG)) {
+ out.setHs2User(in.getOptionValue(HS2_USER_ARG));
+ }
+ if (in.hasOption(HS2_KEYTAB_ARG)) {
+ out.setHs2Keytab(in.getOptionValue(HS2_KEYTAB_ARG));
+ }
}
@@ -1618,6 +1647,8 @@
throw new InvalidOptionsException("Importing to external Hive table requires --hive-import parameter to be set."
+ HELP_STR);
}
+
+ validateHS2Options(options);
}
protected void validateAccumuloOptions(SqoopOptions options)
@@ -1851,5 +1882,26 @@
"Was called with the --direct option, but no direct connector available.");
}
}
-}
+ protected void validateHS2Options(SqoopOptions options) throws SqoopOptions.InvalidOptionsException {
+ final String withoutTemplate = "The %s option cannot be used without the %s option.";
+ final String withTemplate = "The %s option cannot be used with the %s option.";
+
+ if (isSet(options.getHs2Url()) && !options.doHiveImport()) {
+ throw new InvalidOptionsException(format(withoutTemplate, HS2_URL_ARG, HIVE_IMPORT_ARG));
+ }
+
+ if (isSet(options.getHs2User()) && !isSet(options.getHs2Url())) {
+ throw new InvalidOptionsException(format(withoutTemplate, HS2_USER_ARG, HS2_URL_ARG));
+ }
+
+ if (isSet(options.getHs2Keytab()) && !isSet(options.getHs2User())) {
+ throw new InvalidOptionsException(format(withoutTemplate, HS2_KEYTAB_ARG, HS2_USER_ARG));
+ }
+
+ if (isSet(options.getHs2Url()) && (options.getFileLayout() == SqoopOptions.FileLayout.ParquetFile)) {
+ throw new InvalidOptionsException(format(withTemplate, HS2_URL_ARG, FMT_PARQUETFILE_ARG));
+ }
+
+ }
+}
diff --git a/src/java/org/apache/sqoop/tool/CreateHiveTableTool.java b/src/java/org/apache/sqoop/tool/CreateHiveTableTool.java
index d259566..02f5d2d 100644
--- a/src/java/org/apache/sqoop/tool/CreateHiveTableTool.java
+++ b/src/java/org/apache/sqoop/tool/CreateHiveTableTool.java
@@ -30,7 +30,8 @@
import org.apache.sqoop.SqoopOptions.InvalidOptionsException;
import org.apache.sqoop.cli.RelatedOptions;
import org.apache.sqoop.cli.ToolOptions;
-import org.apache.sqoop.hive.HiveImport;
+import org.apache.sqoop.hive.HiveClient;
+import org.apache.sqoop.hive.HiveClientFactory;
/**
* Tool that creates a Hive table definition.
@@ -40,8 +41,15 @@
public static final Log LOG = LogFactory.getLog(
CreateHiveTableTool.class.getName());
- public CreateHiveTableTool() {
+ private final HiveClientFactory hiveClientFactory;
+
+ public CreateHiveTableTool(HiveClientFactory hiveClientFactory) {
super("create-hive-table");
+ this.hiveClientFactory = hiveClientFactory;
+ }
+
+ public CreateHiveTableTool() {
+ this(new HiveClientFactory());
}
@Override
@@ -52,10 +60,8 @@
}
try {
- HiveImport hiveImport = new HiveImport(options, manager,
- options.getConf(), false);
- hiveImport.importTable(options.getTableName(),
- options.getHiveTableName(), true);
+ HiveClient hiveClient = hiveClientFactory.createHiveClient(options, manager);
+ hiveClient.createTable();
} catch (IOException ioe) {
LOG.error("Encountered IOException running create table job: "
+ StringUtils.stringifyException(ioe));
diff --git a/src/java/org/apache/sqoop/tool/ImportAllTablesTool.java b/src/java/org/apache/sqoop/tool/ImportAllTablesTool.java
index 18f7a0a..6fb4a66 100644
--- a/src/java/org/apache/sqoop/tool/ImportAllTablesTool.java
+++ b/src/java/org/apache/sqoop/tool/ImportAllTablesTool.java
@@ -31,7 +31,6 @@
import org.apache.sqoop.SqoopOptions;
import org.apache.sqoop.SqoopOptions.InvalidOptionsException;
import org.apache.sqoop.cli.RelatedOptions;
-import org.apache.sqoop.hive.HiveImport;
import org.apache.sqoop.util.ImportException;
/**
@@ -75,7 +74,6 @@
@Override
/** {@inheritDoc} */
public int run(SqoopOptions options) {
- HiveImport hiveImport = null;
Set<String> excludes = new HashSet<String>();
if (!init(options)) {
@@ -83,9 +81,6 @@
}
try {
- if (options.doHiveImport()) {
- hiveImport = new HiveImport(options, manager, options.getConf(), false);
- }
if (options.getAllTablesExclude() != null) {
excludes.addAll(Arrays.asList(options.getAllTablesExclude().split(",")));
@@ -102,7 +97,8 @@
System.out.println("Skipping table: " + tableName);
} else {
SqoopOptions clonedOptions = (SqoopOptions) options.clone();
- importTable(clonedOptions, tableName, hiveImport);
+ clonedOptions.setTableName(tableName);
+ importTable(clonedOptions);
}
}
}
diff --git a/src/java/org/apache/sqoop/tool/ImportTool.java b/src/java/org/apache/sqoop/tool/ImportTool.java
index e992005..ee79d8b 100644
--- a/src/java/org/apache/sqoop/tool/ImportTool.java
+++ b/src/java/org/apache/sqoop/tool/ImportTool.java
@@ -42,7 +42,8 @@
import org.apache.sqoop.SqoopOptions.InvalidOptionsException;
import org.apache.sqoop.cli.RelatedOptions;
import org.apache.sqoop.cli.ToolOptions;
-import org.apache.sqoop.hive.HiveImport;
+import org.apache.sqoop.hive.HiveClient;
+import org.apache.sqoop.hive.HiveClientFactory;
import org.apache.sqoop.manager.ImportJobContext;
import org.apache.sqoop.mapreduce.MergeJob;
import org.apache.sqoop.metastore.JobData;
@@ -78,18 +79,21 @@
// Set classloader for local job runner
private ClassLoader prevClassLoader = null;
+ private final HiveClientFactory hiveClientFactory;
+
public ImportTool() {
this("import", false);
}
public ImportTool(String toolName, boolean allTables) {
- this(toolName, new CodeGenTool(), allTables);
+ this(toolName, new CodeGenTool(), allTables, new HiveClientFactory());
}
- public ImportTool(String toolName, CodeGenTool codeGenerator, boolean allTables) {
+ public ImportTool(String toolName, CodeGenTool codeGenerator, boolean allTables, HiveClientFactory hiveClientFactory) {
super(toolName);
this.codeGenerator = codeGenerator;
this.allTables = allTables;
+ this.hiveClientFactory = hiveClientFactory;
}
@Override
@@ -499,17 +503,16 @@
* Import a table or query.
* @return true if an import was performed, false otherwise.
*/
- protected boolean importTable(SqoopOptions options, String tableName,
- HiveImport hiveImport) throws IOException, ImportException {
+ protected boolean importTable(SqoopOptions options) throws IOException, ImportException {
String jarFile = null;
// Generate the ORM code for the tables.
- jarFile = codeGenerator.generateORM(options, tableName);
+ jarFile = codeGenerator.generateORM(options, options.getTableName());
- Path outputPath = getOutputPath(options, tableName);
+ Path outputPath = getOutputPath(options, options.getTableName());
// Do the actual import.
- ImportJobContext context = new ImportJobContext(tableName, jarFile,
+ ImportJobContext context = new ImportJobContext(options.getTableName(), jarFile,
options, outputPath);
// If we're doing an incremental import, set up the
@@ -522,7 +525,7 @@
deleteTargetDir(context);
}
- if (null != tableName) {
+ if (null != options.getTableName()) {
manager.importTable(context);
} else {
manager.importQuery(context);
@@ -540,7 +543,8 @@
// For Parquet file, the import action will create hive table directly via
// kite. So there is no need to do hive import as a post step again.
if (options.getFileLayout() != SqoopOptions.FileLayout.ParquetFile) {
- hiveImport.importTable(tableName, options.getHiveTableName(), false);
+ HiveClient hiveClient = hiveClientFactory.createHiveClient(options, manager);
+ hiveClient.importTable();
}
}
@@ -609,8 +613,6 @@
@Override
/** {@inheritDoc} */
public int run(SqoopOptions options) {
- HiveImport hiveImport = null;
-
if (allTables) {
// We got into this method, but we should be in a subclass.
// (This method only handles a single table)
@@ -626,12 +628,8 @@
codeGenerator.setManager(manager);
try {
- if (options.doHiveImport()) {
- hiveImport = new HiveImport(options, manager, options.getConf(), false);
- }
-
// Import a single table (or query) the user specified.
- importTable(options, options.getTableName(), hiveImport);
+ importTable(options);
} catch (IllegalArgumentException iea) {
LOG.error(IMPORT_FAILED_ERROR_MSG + iea.getMessage());
rethrowIfRequired(options, iea);
diff --git a/src/test/org/apache/sqoop/hive/HiveServer2ConnectionFactoryInitializerTest.java b/src/test/org/apache/sqoop/hive/HiveServer2ConnectionFactoryInitializerTest.java
new file mode 100644
index 0000000..4d2cb2f
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/HiveServer2ConnectionFactoryInitializerTest.java
@@ -0,0 +1,117 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+import org.apache.sqoop.db.decorator.KerberizedConnectionFactoryDecorator;
+import org.assertj.core.api.SoftAssertions;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class HiveServer2ConnectionFactoryInitializerTest {
+
+ private static final String TEST_HS2_URL = "jdbc:hive2://myhost:10000/default";
+
+ private static final String TEST_HS2_USER = "testuser";
+
+ private static final String TEST_HS2_KEYTAB = "testkeytab";
+
+ private HiveServer2ConnectionFactoryInitializer connectionFactoryInitializer;
+
+ private SqoopOptions sqoopOptions;
+
+ private Configuration configuration;
+
+ private SoftAssertions softly;
+
+ @Before
+ public void before() {
+ connectionFactoryInitializer = new HiveServer2ConnectionFactoryInitializer();
+ sqoopOptions = mock(SqoopOptions.class);
+ configuration = mock(Configuration.class);
+ softly = new SoftAssertions();
+
+ when(sqoopOptions.getHs2User()).thenReturn(TEST_HS2_USER);
+ when(sqoopOptions.getConf()).thenReturn(configuration);
+ }
+
+ @Test
+ public void testCreateJdbcConnectionFactoryWithoutKerberosConfiguredReturnsHiveServer2ConnectionFactory() throws Exception {
+ JdbcConnectionFactory connectionFactory = connectionFactoryInitializer.createJdbcConnectionFactory(sqoopOptions);
+
+ assertThat(connectionFactory, instanceOf(HiveServer2ConnectionFactory.class));
+ }
+
+ @Test
+ public void testCreateJdbcConnectionFactoryInitializesConnectionStringProperly() throws Exception {
+ when(sqoopOptions.getHs2Url()).thenReturn(TEST_HS2_URL);
+ HiveServer2ConnectionFactory connectionFactory = (HiveServer2ConnectionFactory) connectionFactoryInitializer.createJdbcConnectionFactory(sqoopOptions);
+
+ assertEquals(TEST_HS2_URL, connectionFactory.getConnectionString());
+ }
+
+ @Test
+ public void testCreateJdbcConnectionFactoryInitializesConnectionUsernameProperly() throws Exception {
+ HiveServer2ConnectionFactory connectionFactory = (HiveServer2ConnectionFactory) connectionFactoryInitializer.createJdbcConnectionFactory(sqoopOptions);
+
+ assertEquals(TEST_HS2_USER, connectionFactory.getUsername());
+ }
+
+ @Test
+ public void testCreateJdbcConnectionFactoryWithoutHs2UserSpecifiedInitializesConnectionUsernameProperly() throws Exception {
+ when(sqoopOptions.getHs2User()).thenReturn(null);
+ String expectedUsername = UserGroupInformation.getLoginUser().getUserName();
+ HiveServer2ConnectionFactory connectionFactory = (HiveServer2ConnectionFactory) connectionFactoryInitializer.createJdbcConnectionFactory(sqoopOptions);
+
+ assertEquals(expectedUsername, connectionFactory.getUsername());
+ }
+
+ @Test
+ public void testCreateJdbcConnectionFactoryWithKerberosConfiguredReturnsKerberizedConnectionFactoryDecorator() throws Exception {
+ when(sqoopOptions.getHs2Keytab()).thenReturn(TEST_HS2_KEYTAB);
+
+ JdbcConnectionFactory connectionFactory = connectionFactoryInitializer.createJdbcConnectionFactory(sqoopOptions);
+
+ assertThat(connectionFactory, instanceOf(KerberizedConnectionFactoryDecorator.class));
+ }
+
+ @Test
+ public void testCreateJdbcConnectionFactoryWithKerberosConfiguredInitializesDecoratorProperly() throws Exception {
+ when(sqoopOptions.getHs2Keytab()).thenReturn(TEST_HS2_KEYTAB);
+
+ KerberizedConnectionFactoryDecorator connectionFactory = (KerberizedConnectionFactoryDecorator) connectionFactoryInitializer.createJdbcConnectionFactory(sqoopOptions);
+
+ softly.assertThat(connectionFactory.getDecorated()).isInstanceOf(HiveServer2ConnectionFactory.class);
+ softly.assertThat(connectionFactory.getAuthenticator().getConfiguration()).isSameAs(configuration);
+ softly.assertThat(connectionFactory.getAuthenticator().getPrincipal()).isEqualTo(TEST_HS2_USER);
+ softly.assertThat(connectionFactory.getAuthenticator().getKeytabLocation()).isEqualTo(TEST_HS2_KEYTAB);
+
+ softly.assertAll();
+ }
+
+}
\ No newline at end of file
diff --git a/src/test/org/apache/sqoop/hive/TestHiveClientFactory.java b/src/test/org/apache/sqoop/hive/TestHiveClientFactory.java
new file mode 100644
index 0000000..a3c2dc9
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/TestHiveClientFactory.java
@@ -0,0 +1,115 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+import org.apache.sqoop.manager.ConnManager;
+import org.assertj.core.api.SoftAssertions;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestHiveClientFactory {
+
+ private static final String TEST_HS2_URL = "jdbc:hive2://myhost:10000/default";
+
+ private static final String TEST_TABLE_NAME = "testTableName";
+
+ private static final String TEST_HIVE_TABLE_NAME = "testHiveTableName";
+
+ private HiveClientFactory hiveClientFactory;
+
+ private ConnManager connectionManager;
+
+ private SqoopOptions sqoopOptions;
+
+ private Configuration configuration;
+
+ private JdbcConnectionFactory jdbcConnectionFactory;
+
+ private HiveServer2ConnectionFactoryInitializer connectionFactoryInitializer;
+
+ private SoftAssertions softly;
+
+ @Before
+ public void before() {
+ connectionFactoryInitializer = mock(HiveServer2ConnectionFactoryInitializer.class);
+ hiveClientFactory = new HiveClientFactory(connectionFactoryInitializer);
+ softly = new SoftAssertions();
+
+ connectionManager = mock(ConnManager.class);
+ sqoopOptions = mock(SqoopOptions.class);
+ configuration = mock(Configuration.class);
+ jdbcConnectionFactory = mock(JdbcConnectionFactory.class);
+
+ when(sqoopOptions.getConf()).thenReturn(configuration);
+ when(sqoopOptions.getTableName()).thenReturn(TEST_TABLE_NAME);
+ when(sqoopOptions.getHiveTableName()).thenReturn(TEST_HIVE_TABLE_NAME);
+ }
+
+ @Test
+ public void testCreateHiveClientCreatesHiveImportWhenHs2UrlIsNotProvided() throws Exception {
+ HiveClient hiveClient = hiveClientFactory.createHiveClient(sqoopOptions, connectionManager);
+ assertThat(hiveClient, instanceOf(HiveImport.class));
+ }
+
+ @Test
+ public void testCreateHiveClientInitializesHiveImportProperly() throws Exception {
+ HiveImport hiveImport = (HiveImport) hiveClientFactory.createHiveClient(sqoopOptions, connectionManager);
+
+ softly.assertThat(hiveImport.getOptions()).isSameAs(sqoopOptions);
+ softly.assertThat(hiveImport.getConnManager()).isSameAs(connectionManager);
+ softly.assertThat(hiveImport.getConfiguration()).isSameAs(configuration);
+ softly.assertThat(hiveImport.isGenerateOnly()).isFalse();
+ softly.assertAll();
+ }
+
+ @Test
+ public void testCreateHiveClientCreatesHiveServer2ClientWhenHs2UrlIsProvided() throws Exception {
+ when(sqoopOptions.getHs2Url()).thenReturn(TEST_HS2_URL);
+ HiveClient hiveClient = hiveClientFactory.createHiveClient(sqoopOptions, connectionManager);
+ assertThat(hiveClient, instanceOf(HiveServer2Client.class));
+ }
+
+ @Test
+ public void testCreateHiveClientInitializesHiveServer2ClientProperly() throws Exception {
+ when(sqoopOptions.getHs2Url()).thenReturn(TEST_HS2_URL);
+ when(connectionFactoryInitializer.createJdbcConnectionFactory(sqoopOptions)).thenReturn(jdbcConnectionFactory);
+
+ HiveServer2Client hs2Client = (HiveServer2Client) hiveClientFactory.createHiveClient(sqoopOptions, connectionManager);
+
+ softly.assertThat(hs2Client.getSqoopOptions()).isSameAs(sqoopOptions);
+ softly.assertThat(hs2Client.getHs2ConnectionFactory()).isSameAs(jdbcConnectionFactory);
+ softly.assertThat(hs2Client.getTableDefWriter().getOptions()).isSameAs(sqoopOptions);
+ softly.assertThat(hs2Client.getTableDefWriter().getConnManager()).isSameAs(connectionManager);
+ softly.assertThat(hs2Client.getTableDefWriter().getInputTableName()).isEqualTo(TEST_TABLE_NAME);
+ softly.assertThat(hs2Client.getTableDefWriter().getOutputTableName()).isEqualTo(TEST_HIVE_TABLE_NAME);
+ softly.assertThat(hs2Client.getTableDefWriter().getConfiguration()).isSameAs(configuration);
+ softly.assertThat(hs2Client.getTableDefWriter().isCommentsEnabled()).isFalse();
+
+ softly.assertAll();
+ }
+
+}
\ No newline at end of file
diff --git a/src/test/org/apache/sqoop/hive/TestHiveMiniCluster.java b/src/test/org/apache/sqoop/hive/TestHiveMiniCluster.java
new file mode 100644
index 0000000..419f888
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/TestHiveMiniCluster.java
@@ -0,0 +1,124 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+import org.apache.sqoop.hive.minicluster.AuthenticationConfiguration;
+import org.apache.sqoop.hive.minicluster.HiveMiniCluster;
+import org.apache.sqoop.hive.minicluster.KerberosAuthenticationConfiguration;
+import org.apache.sqoop.hive.minicluster.NoAuthenticationConfiguration;
+import org.apache.sqoop.hive.minicluster.PasswordAuthenticationConfiguration;
+import org.apache.sqoop.infrastructure.kerberos.MiniKdcInfrastructureRule;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Arrays;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(Parameterized.class)
+public class TestHiveMiniCluster {
+
+ @ClassRule
+ public static MiniKdcInfrastructureRule miniKdcInfrastructure = new MiniKdcInfrastructureRule();
+
+ private static final String TEST_USERNAME = "sqoop";
+
+ private static final String TEST_PASSWORD = "secret";
+
+ @Parameters(name = "config = {0}")
+ public static Iterable<? extends Object> authenticationParameters() {
+ return Arrays.asList(new NoAuthenticationConfiguration(),
+ new PasswordAuthenticationConfiguration(TEST_USERNAME, TEST_PASSWORD),
+ new KerberosAuthenticationConfiguration(miniKdcInfrastructure));
+ }
+
+ private static final String CREATE_TABLE_SQL = "CREATE TABLE TestTable (id int)";
+
+ private static final String INSERT_SQL = "INSERT INTO TestTable VALUES (?)";
+
+ private static final String SELECT_SQL = "SELECT * FROM TestTable";
+
+ private static final int TEST_VALUE = 10;
+
+ private final AuthenticationConfiguration authenticationConfiguration;
+
+ private HiveMiniCluster hiveMiniCluster;
+
+ private JdbcConnectionFactory connectionFactory;
+
+ public TestHiveMiniCluster(AuthenticationConfiguration authenticationConfiguration) {
+ this.authenticationConfiguration = authenticationConfiguration;
+ }
+
+ @Before
+ public void before() throws SQLException {
+ hiveMiniCluster = new HiveMiniCluster(authenticationConfiguration);
+ hiveMiniCluster.start();
+
+ connectionFactory = authenticationConfiguration.decorateConnectionFactory(new HiveServer2ConnectionFactory(hiveMiniCluster.getUrl(), TEST_USERNAME, TEST_PASSWORD));
+ }
+
+ @Test
+ public void testInsertedRowCanBeReadFromTable() throws Exception {
+ createTestTable();
+ insertRowIntoTestTable();
+
+ assertEquals(TEST_VALUE, getDataFromTestTable());
+ }
+
+ private void insertRowIntoTestTable() throws SQLException {
+ try (Connection conn = connectionFactory.createConnection(); PreparedStatement stmnt = conn.prepareStatement(INSERT_SQL)) {
+ stmnt.setInt(1, TEST_VALUE);
+ stmnt.executeUpdate();
+ }
+ }
+
+ private int getDataFromTestTable() throws SQLException {
+ try (Connection conn = connectionFactory.createConnection(); PreparedStatement stmnt = conn.prepareStatement(SELECT_SQL)) {
+ ResultSet resultSet = stmnt.executeQuery();
+ resultSet.next();
+ return resultSet.getInt(1);
+ }
+ }
+
+ private void createTestTable() throws SQLException {
+ try (Connection conn = connectionFactory.createConnection(); PreparedStatement stmnt = conn.prepareStatement(CREATE_TABLE_SQL)) {
+ stmnt.executeUpdate();
+ }
+ }
+
+ @After
+ public void after() {
+ hiveMiniCluster.stop();
+ UserGroupInformation.setConfiguration(new Configuration());
+ }
+
+}
diff --git a/src/test/org/apache/sqoop/hive/TestHiveServer2Client.java b/src/test/org/apache/sqoop/hive/TestHiveServer2Client.java
new file mode 100644
index 0000000..0261729
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/TestHiveServer2Client.java
@@ -0,0 +1,213 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+
+import static java.util.Arrays.asList;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.mockito.Matchers.anyList;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class TestHiveServer2Client {
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ private static final String CREATE_TABLE_STATEMENT = "createTableStatement";
+
+ private static final String LOAD_DATA_STATEMENT = "loadDataStatement";
+
+ private static final List<String> TEST_COMMANDS = asList("command1", "command2", "command3");
+
+ private HiveServer2Client hs2Client;
+
+ private HiveServer2Client hs2ClientSpy;
+
+ private SqoopOptions sqoopOptions;
+
+ private TableDefWriter tableDefWriter;
+
+ private JdbcConnectionFactory hs2ConnectionFactory;
+
+ private Connection hs2Connection;
+
+ private PreparedStatement preparedStatement;
+
+ private HiveClientCommon hiveClientCommon;
+
+ private Path finalPath;
+
+ private Configuration configuration;
+
+ @Before
+ public void before() throws Exception {
+ sqoopOptions = mock(SqoopOptions.class);
+ tableDefWriter = mock(TableDefWriter.class);
+ hs2ConnectionFactory = mock(JdbcConnectionFactory.class);
+ hs2Connection = mock(Connection.class);
+ preparedStatement = mock(PreparedStatement.class);
+ hiveClientCommon = mock(HiveClientCommon.class);
+ finalPath = mock(Path.class);
+ configuration = mock(Configuration.class);
+
+ when(sqoopOptions.getConf()).thenReturn(configuration);
+ when(hs2ConnectionFactory.createConnection()).thenReturn(hs2Connection);
+ when(hs2Connection.prepareStatement(anyString())).thenReturn(preparedStatement);
+
+ when(tableDefWriter.getCreateTableStmt()).thenReturn(CREATE_TABLE_STATEMENT);
+ when(tableDefWriter.getLoadDataStmt()).thenReturn(LOAD_DATA_STATEMENT);
+ when(tableDefWriter.getFinalPath()).thenReturn(finalPath);
+
+ hs2Client = new HiveServer2Client(sqoopOptions, tableDefWriter, hs2ConnectionFactory, hiveClientCommon);
+ hs2ClientSpy = spy(hs2Client);
+ }
+
+ @Test
+ public void testImportTableExecutesHiveImportWithCreateTableAndLoadDataCommands() throws Exception {
+ doNothing().when(hs2ClientSpy).executeHiveImport(anyList());
+
+ hs2ClientSpy.importTable();
+
+ verify(hs2ClientSpy, times(1)).executeHiveImport(asList(CREATE_TABLE_STATEMENT, LOAD_DATA_STATEMENT));
+ }
+
+ @Test
+ public void testCreateTableExecutesHiveImportWithCreateTableCommandOnly() throws Exception {
+ doNothing().when(hs2ClientSpy).executeHiveImport(anyList());
+
+ hs2ClientSpy.createTable();
+
+ verify(hs2ClientSpy, times(1)).executeHiveImport(asList(CREATE_TABLE_STATEMENT));
+ }
+
+ @Test
+ public void testExecuteHiveImportInvokesMethodsInCorrectSequence() throws Exception {
+ InOrder inOrder = Mockito.inOrder(hiveClientCommon, hs2ClientSpy);
+ doNothing().when(hs2ClientSpy).executeCommands(TEST_COMMANDS);
+
+ hs2ClientSpy.executeHiveImport(TEST_COMMANDS);
+
+ inOrder.verify(hiveClientCommon).removeTempLogs(configuration, finalPath);
+ inOrder.verify(hiveClientCommon).indexLzoFiles(sqoopOptions, finalPath);
+ inOrder.verify(hs2ClientSpy).executeCommands(TEST_COMMANDS);
+ inOrder.verify(hiveClientCommon).cleanUp(configuration, finalPath);
+ }
+
+ @Test
+ public void testExecuteHiveImportThrowsRuntimeExceptionWhenExecuteCommandsThrows() throws Exception {
+ SQLException sqlException = mock(SQLException.class);
+ doThrow(sqlException).when(hs2ClientSpy).executeCommands(anyList());
+
+ expectedException.expect(RuntimeException.class);
+ expectedException.expectMessage("Error executing Hive import.");
+
+ hs2ClientSpy.executeHiveImport(TEST_COMMANDS);
+ }
+
+ @Test
+ public void testExecuteCommandsCreatesExactlyOneConnection() throws Exception {
+ hs2Client.executeCommands(TEST_COMMANDS);
+
+ verify(hs2ConnectionFactory, times(1)).createConnection();
+ }
+
+ @Test
+ public void testExecuteCommandsClosesConnectionWhenStatementExecutionIsSuccessful() throws Exception {
+ hs2Client.executeCommands(TEST_COMMANDS);
+
+ verify(hs2Connection).close();
+ }
+
+ @Test
+ public void testExecuteCommandsClosesConnectionWhenStatementExecutionThrows() throws Exception {
+ when(hs2Connection.prepareStatement(anyString())).thenThrow(new SQLException());
+
+ expectedException.expect(SQLException.class);
+ hs2Client.executeCommands(TEST_COMMANDS);
+
+ verify(hs2Connection).close();
+ }
+
+ @Test
+ public void testExecuteCommandsClosesPreparedStatementsWhenStatementExecutionIsSuccessful() throws Exception {
+ hs2Client.executeCommands(TEST_COMMANDS);
+
+ verify(preparedStatement, times(TEST_COMMANDS.size())).close();
+ }
+
+ @Test
+ public void testExecuteCommandsClosesPreparedStatementWhenStatementExecutionThrows() throws Exception {
+ when(preparedStatement.execute()).thenThrow(new SQLException());
+
+ expectedException.expect(SQLException.class);
+ hs2Client.executeCommands(TEST_COMMANDS);
+
+ verify(preparedStatement).close();
+ }
+
+ @Test
+ public void testExecuteCommandsThrowsWhenCreateConnectionThrows() throws Exception {
+ RuntimeException expected = mock(RuntimeException.class);
+ when(hs2ConnectionFactory.createConnection()).thenThrow(expected);
+
+ expectedException.expect(equalTo(expected));
+ hs2Client.executeCommands(TEST_COMMANDS);
+ }
+
+ @Test
+ public void testExecuteCommandsThrowsWhenPrepareStatementThrows() throws Exception {
+ SQLException expected = mock(SQLException.class);
+ when(hs2Connection.prepareStatement(anyString())).thenThrow(expected);
+
+ expectedException.expect(equalTo(expected));
+ hs2Client.executeCommands(TEST_COMMANDS);
+ }
+
+ @Test
+ public void testExecuteCommandsThrowsWhenExecuteStatementThrows() throws Exception {
+ SQLException expected = mock(SQLException.class);
+ when(preparedStatement.execute()).thenThrow(expected);
+
+ expectedException.expect(equalTo(expected));
+ hs2Client.executeCommands(TEST_COMMANDS);
+ }
+
+}
diff --git a/src/test/org/apache/sqoop/hive/TestHiveServer2TextImport.java b/src/test/org/apache/sqoop/hive/TestHiveServer2TextImport.java
new file mode 100644
index 0000000..f6d591b
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/TestHiveServer2TextImport.java
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.sqoop.hive.minicluster.HiveMiniCluster;
+import org.apache.sqoop.hive.minicluster.KerberosAuthenticationConfiguration;
+import org.apache.sqoop.infrastructure.kerberos.MiniKdcInfrastructureRule;
+import org.apache.sqoop.testutil.ArgumentArrayBuilder;
+import org.apache.sqoop.testutil.HiveServer2TestUtil;
+import org.apache.sqoop.testutil.ImportJobTestCase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestHiveServer2TextImport extends ImportJobTestCase {
+
+ @ClassRule
+ public static MiniKdcInfrastructureRule miniKdcInfrastructure = new MiniKdcInfrastructureRule();
+
+ private HiveMiniCluster hiveMiniCluster;
+
+ private HiveServer2TestUtil hiveServer2TestUtil;
+
+ @Override
+ @Before
+ public void setUp() {
+ super.setUp();
+ KerberosAuthenticationConfiguration authenticationConfiguration = new KerberosAuthenticationConfiguration(miniKdcInfrastructure);
+ hiveMiniCluster = new HiveMiniCluster(authenticationConfiguration);
+ hiveMiniCluster.start();
+ hiveServer2TestUtil = new HiveServer2TestUtil(hiveMiniCluster.getUrl());
+ }
+
+ @Override
+ @After
+ public void tearDown() {
+ super.tearDown();
+ hiveMiniCluster.stop();
+ }
+
+ @Test
+ public void testImport() throws Exception {
+ List<Object> columnValues = Arrays.<Object>asList("test", 42, "somestring");
+
+ String[] types = {"VARCHAR(32)", "INTEGER", "CHAR(64)"};
+ createTableWithColTypes(types, toStringArray(columnValues));
+
+ String[] args = new ArgumentArrayBuilder()
+ .withProperty(YarnConfiguration.RM_PRINCIPAL, miniKdcInfrastructure.getTestPrincipal())
+ .withOption("connect", getConnectString())
+ .withOption("table", getTableName())
+ .withOption("hive-import")
+ .withOption("hs2-url", hiveMiniCluster.getUrl())
+ .withOption("split-by", getColName(1))
+ .build();
+
+ runImport(args);
+
+ List<List<Object>> rows = hiveServer2TestUtil.loadRawRowsFromTable(getTableName());
+ assertEquals(columnValues, rows.get(0));
+ }
+
+ private String[] toStringArray(List<Object> columnValues) {
+ String[] result = new String[columnValues.size()];
+
+ for (int i = 0; i < columnValues.size(); i++) {
+ if (columnValues.get(i) instanceof String) {
+ result[i] = StringUtils.wrap((String) columnValues.get(i), '\'');
+ } else {
+ result[i] = columnValues.get(i).toString();
+ }
+ }
+
+ return result;
+ }
+
+}
diff --git a/src/test/org/apache/sqoop/hive/TestTableDefWriter.java b/src/test/org/apache/sqoop/hive/TestTableDefWriter.java
index 8bdc3be..3ea61f6 100644
--- a/src/test/org/apache/sqoop/hive/TestTableDefWriter.java
+++ b/src/test/org/apache/sqoop/hive/TestTableDefWriter.java
@@ -42,6 +42,7 @@
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -248,6 +249,13 @@
assertTrue(createTable.contains("`db`.`outputTable`"));
}
+ @Test
+ public void testGetCreateTableStmtDiscardsConnection() throws Exception {
+ writer.getCreateTableStmt();
+
+ verify(connManager).discardConnection(true);
+ }
+
private void setUpMockConnManager(String tableName, Map<String, Integer> typeMap) {
when(connManager.getColumnTypes(tableName)).thenReturn(typeMap);
when(connManager.getColumnNames(tableName)).thenReturn(typeMap.keySet().toArray(new String[]{}));
diff --git a/src/test/org/apache/sqoop/hive/minicluster/AuthenticationConfiguration.java b/src/test/org/apache/sqoop/hive/minicluster/AuthenticationConfiguration.java
new file mode 100644
index 0000000..4171337
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/minicluster/AuthenticationConfiguration.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive.minicluster;
+
+import org.apache.sqoop.db.JdbcConnectionFactory;
+
+import java.security.PrivilegedAction;
+import java.util.Map;
+
+public interface AuthenticationConfiguration {
+
+ Map<String, String> getAuthenticationConfig();
+
+ String getUrlParams();
+
+ <T> T doAsAuthenticated(PrivilegedAction<T> action);
+
+ void init();
+
+ JdbcConnectionFactory decorateConnectionFactory(JdbcConnectionFactory connectionFactory);
+
+}
diff --git a/src/test/org/apache/sqoop/hive/minicluster/HiveMiniCluster.java b/src/test/org/apache/sqoop/hive/minicluster/HiveMiniCluster.java
new file mode 100644
index 0000000..19bb760
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/minicluster/HiveMiniCluster.java
@@ -0,0 +1,175 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive.minicluster;
+
+import com.google.common.io.Files;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hive.service.Service;
+import org.apache.hive.service.server.HiveServer2;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.security.PrivilegedAction;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+
+public class HiveMiniCluster {
+
+ private static final Log LOG = LogFactory.getLog(HiveMiniCluster.class.getName());
+
+ private static final String DEFAULT_HOST = "127.0.0.1";
+
+ private static final int DEFAULT_PORT = 10000;
+
+ private final String hostName;
+
+ private final int port;
+
+ private final String tempFolderPath;
+
+ private final AuthenticationConfiguration authenticationConfiguration;
+
+ private final HiveServer2 hiveServer2;
+
+ private HiveConf config;
+
+ public HiveMiniCluster(AuthenticationConfiguration authenticationConfiguration) {
+ this(DEFAULT_HOST, DEFAULT_PORT, authenticationConfiguration);
+ }
+
+ public HiveMiniCluster(String hostname, int port, AuthenticationConfiguration authenticationConfiguration) {
+ this(hostname, port, Files.createTempDir().getAbsolutePath(), authenticationConfiguration);
+ }
+
+ public HiveMiniCluster(String hostname, int port, String tempFolderPath, AuthenticationConfiguration authenticationConfiguration) {
+ this.hostName = hostname;
+ this.port = port;
+ this.tempFolderPath = tempFolderPath;
+ this.authenticationConfiguration = authenticationConfiguration;
+ this.hiveServer2 = new HiveServer2();
+ }
+
+ private void createHiveConf() {
+ config = new HiveConf();
+ config.set(HiveConf.ConfVars.METASTOREWAREHOUSE.varname, tempFolderPath);
+ config.set(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_BIND_HOST.varname, getHostName());
+ config.setInt(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_PORT.varname, getPort());
+ config.set(HiveConf.ConfVars.METASTORECONNECTURLKEY.varname, getMetastoreConnectUrl());
+
+ for (Map.Entry<String, String> authConfig : authenticationConfiguration.getAuthenticationConfig().entrySet()) {
+ config.set(authConfig.getKey(), authConfig.getValue());
+ }
+ }
+
+ public void start() {
+ try {
+ authenticationConfiguration.init();
+ createHiveConf();
+ createHiveSiteXml();
+ startHiveServer();
+ waitForStartUp();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void createHiveSiteXml() throws IOException {
+ File hiveSiteXmlFile = new File(tempFolderPath,"hive-site.xml");
+ try (OutputStream out = new FileOutputStream(hiveSiteXmlFile)) {
+ config.writeXml(out);
+ }
+
+ HiveConf.setHiveSiteLocation(hiveSiteXmlFile.toURI().toURL());
+ }
+
+ private void startHiveServer() throws Exception {
+ authenticationConfiguration.doAsAuthenticated(new PrivilegedAction<Void>() {
+ @Override
+ public Void run() {
+ hiveServer2.init(config);
+ hiveServer2.start();
+ return null;
+ }
+ });
+ }
+
+ public void stop() {
+ hiveServer2.stop();
+ HiveConf.setHiveSiteLocation(null);
+ try {
+ FileUtils.deleteDirectory(new File(tempFolderPath));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public HiveConf getConfig() {
+ return config;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public String getHostName() {
+ return hostName;
+ }
+
+ public String getUrl() {
+ return String.format("jdbc:hive2://%s:%d/default%s", hostName, port, authenticationConfiguration.getUrlParams());
+ }
+
+ public String getTempFolderPath() {
+ return tempFolderPath;
+ }
+
+ public String getMetastoreConnectUrl() {
+ return String.format("jdbc:derby:;databaseName=%s/minicluster_metastore_db;create=true", tempFolderPath);
+ }
+
+ public boolean isStarted() {
+ return hiveServer2.getServiceState() == Service.STATE.STARTED;
+ }
+
+ private void waitForStartUp() throws InterruptedException, TimeoutException {
+ final int numberOfAttempts = 500;
+ final long sleepTime = 100;
+ for (int i = 0; i < numberOfAttempts; ++i) {
+ try {
+ LOG.debug("Attempt " + (i + 1) + " to access " + hostName + ":" + port);
+ new Socket(InetAddress.getByName(hostName), port).close();
+ return;
+ } catch (RuntimeException | IOException e) {
+ LOG.debug("Failed to connect to " + hostName + ":" + port, e);
+ }
+
+ Thread.sleep(sleepTime);
+ }
+
+ throw new RuntimeException("Couldn't access new server: " + hostName + ":" + port);
+ }
+
+}
diff --git a/src/test/org/apache/sqoop/hive/minicluster/KerberosAuthenticationConfiguration.java b/src/test/org/apache/sqoop/hive/minicluster/KerberosAuthenticationConfiguration.java
new file mode 100644
index 0000000..549a8c6
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/minicluster/KerberosAuthenticationConfiguration.java
@@ -0,0 +1,86 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive.minicluster;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hive.service.auth.HiveAuthFactory;
+import org.apache.sqoop.authentication.KerberosAuthenticator;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+import org.apache.sqoop.db.decorator.KerberizedConnectionFactoryDecorator;
+import org.apache.sqoop.infrastructure.kerberos.KerberosConfigurationProvider;
+
+import java.security.PrivilegedAction;
+import java.util.HashMap;
+import java.util.Map;
+
+public class KerberosAuthenticationConfiguration implements AuthenticationConfiguration {
+
+ private final KerberosConfigurationProvider kerberosConfig;
+
+ private KerberosAuthenticator authenticator;
+
+ public KerberosAuthenticationConfiguration(KerberosConfigurationProvider kerberosConfig) {
+ this.kerberosConfig = kerberosConfig;
+ }
+
+ @Override
+ public Map<String, String> getAuthenticationConfig() {
+ Map<String, String> result = new HashMap<>();
+
+ result.put(HiveConf.ConfVars.HIVE_SERVER2_KERBEROS_PRINCIPAL.varname, kerberosConfig.getTestPrincipal());
+ result.put(HiveConf.ConfVars.HIVE_SERVER2_KERBEROS_KEYTAB.varname, kerberosConfig.getKeytabFilePath());
+ result.put(HiveConf.ConfVars.HIVE_SERVER2_AUTHENTICATION.varname, HiveAuthFactory.AuthTypes.KERBEROS.toString());
+ result.put(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION, HiveAuthFactory.AuthTypes.KERBEROS.toString());
+ result.put(YarnConfiguration.RM_PRINCIPAL, kerberosConfig.getTestPrincipal());
+
+ return result;
+ }
+
+ @Override
+ public String getUrlParams() {
+ return ";principal=" + kerberosConfig.getTestPrincipal();
+ }
+
+ @Override
+ public <T> T doAsAuthenticated(PrivilegedAction<T> action) {
+ return authenticator.authenticate().doAs(action);
+ }
+
+ @Override
+ public void init() {
+ authenticator = createKerberosAuthenticator();
+ }
+
+ @Override
+ public JdbcConnectionFactory decorateConnectionFactory(JdbcConnectionFactory connectionFactory) {
+ return new KerberizedConnectionFactoryDecorator(connectionFactory, authenticator);
+ }
+
+ private KerberosAuthenticator createKerberosAuthenticator() {
+ Configuration conf = new Configuration();
+ conf.set(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION, "kerberos");
+ KerberosAuthenticator result = new KerberosAuthenticator(conf, kerberosConfig.getTestPrincipal(), kerberosConfig.getKeytabFilePath());
+ return result;
+ }
+
+}
diff --git a/src/test/org/apache/sqoop/hive/minicluster/NoAuthenticationConfiguration.java b/src/test/org/apache/sqoop/hive/minicluster/NoAuthenticationConfiguration.java
new file mode 100644
index 0000000..20502c9
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/minicluster/NoAuthenticationConfiguration.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive.minicluster;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+
+import java.security.PrivilegedAction;
+import java.util.Collections;
+import java.util.Map;
+
+public class NoAuthenticationConfiguration implements AuthenticationConfiguration {
+ @Override
+ public Map<String, String> getAuthenticationConfig() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public String getUrlParams() {
+ return StringUtils.EMPTY;
+ }
+
+ @Override
+ public <T> T doAsAuthenticated(PrivilegedAction<T> action) {
+ return action.run();
+ }
+
+ @Override
+ public void init() {
+ // do nothing
+ }
+
+ @Override
+ public JdbcConnectionFactory decorateConnectionFactory(JdbcConnectionFactory connectionFactory) {
+ return connectionFactory;
+ }
+}
diff --git a/src/test/org/apache/sqoop/hive/minicluster/PasswordAuthenticationConfiguration.java b/src/test/org/apache/sqoop/hive/minicluster/PasswordAuthenticationConfiguration.java
new file mode 100644
index 0000000..79881f7
--- /dev/null
+++ b/src/test/org/apache/sqoop/hive/minicluster/PasswordAuthenticationConfiguration.java
@@ -0,0 +1,83 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.hive.minicluster;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hive.service.auth.HiveAuthFactory;
+import org.apache.hive.service.auth.PasswdAuthenticationProvider;
+import org.apache.sqoop.db.JdbcConnectionFactory;
+
+import javax.security.sasl.AuthenticationException;
+import java.security.PrivilegedAction;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_SERVER2_AUTHENTICATION;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_SERVER2_CUSTOM_AUTHENTICATION_CLASS;
+
+public class PasswordAuthenticationConfiguration implements AuthenticationConfiguration {
+
+ private static String TEST_USERNAME;
+
+ private static String TEST_PASSWORD;
+
+ private static final class TestPasswordAuthenticationProvider implements PasswdAuthenticationProvider {
+
+ @Override
+ public void Authenticate(String user, String password) throws AuthenticationException {
+ if (!(TEST_USERNAME.equals(user) && TEST_PASSWORD.equals(password))) {
+ throw new AuthenticationException("Authentication failed!");
+ }
+ }
+ }
+
+ public PasswordAuthenticationConfiguration(String testUsername, String testPassword) {
+ TEST_USERNAME = testUsername;
+ TEST_PASSWORD = testPassword;
+ }
+
+ @Override
+ public Map<String, String> getAuthenticationConfig() {
+ Map<String, String> result = new HashMap<>();
+ result.put(HIVE_SERVER2_AUTHENTICATION.varname, HiveAuthFactory.AuthTypes.CUSTOM.getAuthName());
+ result.put(HIVE_SERVER2_CUSTOM_AUTHENTICATION_CLASS.varname, TestPasswordAuthenticationProvider.class.getName());
+
+ return result;
+ }
+
+ @Override
+ public String getUrlParams() {
+ return StringUtils.EMPTY;
+ }
+
+ @Override
+ public <T> T doAsAuthenticated(PrivilegedAction<T> action) {
+ return action.run();
+ }
+
+ @Override
+ public void init() {
+ //do nothing
+ }
+
+ @Override
+ public JdbcConnectionFactory decorateConnectionFactory(JdbcConnectionFactory connectionFactory) {
+ return connectionFactory;
+ }
+}
diff --git a/src/test/org/apache/sqoop/testutil/HiveServer2TestUtil.java b/src/test/org/apache/sqoop/testutil/HiveServer2TestUtil.java
new file mode 100644
index 0000000..7993708
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/HiveServer2TestUtil.java
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.testutil;
+
+import org.apache.sqoop.hive.HiveServer2ConnectionFactory;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+public class HiveServer2TestUtil {
+
+ private static final String SELECT_TABLE_QUERY = "SELECT * FROM %s";
+
+ private HiveServer2ConnectionFactory hs2ConnectionFactory;
+
+ public HiveServer2TestUtil(String url) {
+ this(url, null, null);
+ }
+
+ public HiveServer2TestUtil(String url, String username, String password) {
+ hs2ConnectionFactory = new HiveServer2ConnectionFactory(url, username, password);
+ }
+
+ public List<LinkedHashMap<String, Object>> loadRowsFromTable(String tableName) {
+ List<LinkedHashMap<String, Object>> result = new ArrayList<>();
+ try(Connection connection = hs2ConnectionFactory.createConnection();
+ PreparedStatement query = connection.prepareStatement(String.format(SELECT_TABLE_QUERY, tableName))) {
+
+ ResultSet resultSet = query.executeQuery();
+ ResultSetMetaData metaData = resultSet.getMetaData();
+
+ while (resultSet.next()) {
+ LinkedHashMap<String, Object> row = new LinkedHashMap<>();
+ for (int i = 1; i <= metaData.getColumnCount(); i++) {
+ row.put(metaData.getColumnName(i), resultSet.getObject(i));
+ }
+ result.add(row);
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ return result;
+ }
+
+ public List<List<Object>> loadRawRowsFromTable(String tableName) {
+ List<List<Object>> result = new ArrayList<>();
+ List<LinkedHashMap<String, Object>> rowsWithColumnNames = loadRowsFromTable(tableName);
+
+ for (LinkedHashMap<String, Object> rowWithColumnNames : rowsWithColumnNames) {
+ result.add(new ArrayList<>(rowWithColumnNames.values()));
+ }
+
+ return result;
+ }
+
+}
diff --git a/src/test/org/apache/sqoop/tool/TestHiveServer2OptionValidations.java b/src/test/org/apache/sqoop/tool/TestHiveServer2OptionValidations.java
new file mode 100644
index 0000000..4d3f938
--- /dev/null
+++ b/src/test/org/apache/sqoop/tool/TestHiveServer2OptionValidations.java
@@ -0,0 +1,152 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.sqoop.tool;
+
+import org.apache.sqoop.SqoopOptions;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import java.util.Arrays;
+import java.util.Properties;
+
+import static org.apache.sqoop.SqoopOptions.FileLayout.ParquetFile;
+import static org.apache.sqoop.SqoopOptions.IncrementalMode.None;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+@RunWith(Parameterized.class)
+public class TestHiveServer2OptionValidations {
+
+ @Parameters(name = "sqoopTool = {0}")
+ public static Iterable<? extends Object> parameters() {
+ return Arrays.asList(
+ new ImportTool(),
+ new ImportAllTablesTool(),
+ new CreateHiveTableTool());
+ }
+
+ private static final String TEST_HS2_URL = "test-hs2-url";
+ private static final String TEST_HS2_USER = "test-hs2-user";
+ private static final String TEST_HS2_KEYTAB = "test-hs2-keytab";
+ private static final String TEST_TABLE = "testtable";
+ private static final String TEST_CONNECTION_STRING = "testconnectstring";
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ private final BaseSqoopTool sqoopTool;
+
+ private SqoopOptions sqoopOptions;
+
+ public TestHiveServer2OptionValidations(BaseSqoopTool sqoopTool) {
+ this.sqoopTool = spy(sqoopTool);
+ }
+
+ @Before
+ public void before() {
+ sqoopOptions = mock(SqoopOptions.class);
+ when(sqoopOptions.getTableName()).thenReturn(TEST_TABLE);
+ when(sqoopOptions.getIncrementalMode()).thenReturn(None);
+ when(sqoopOptions.getConnectString()).thenReturn(TEST_CONNECTION_STRING);
+ when(sqoopOptions.getMapColumnHive()).thenReturn(new Properties());
+
+
+ doReturn(0).when(sqoopTool).getDashPosition(any(String[].class));
+ }
+
+ @Test
+ public void testValidateOptionsThrowsWhenHs2UrlIsUsedWithoutHiveImport() throws Exception {
+ expectedException.expect(SqoopOptions.InvalidOptionsException.class);
+ expectedException.expectMessage("The hs2-url option cannot be used without the hive-import option.");
+
+ when(sqoopOptions.doHiveImport()).thenReturn(false);
+ when(sqoopOptions.getHs2Url()).thenReturn(TEST_HS2_URL);
+
+ sqoopTool.validateOptions(sqoopOptions);
+ }
+
+ @Test
+ public void testValidateOptionsThrowsWhenHs2UrlIsUsedWithHCatalogImport() throws Exception {
+ expectedException.expect(SqoopOptions.InvalidOptionsException.class);
+ expectedException.expectMessage("The hs2-url option cannot be used without the hive-import option.");
+
+ when(sqoopOptions.getHCatTableName()).thenReturn(TEST_TABLE);
+ when(sqoopOptions.getHs2Url()).thenReturn(TEST_HS2_URL);
+
+ sqoopTool.validateOptions(sqoopOptions);
+ }
+
+ @Test
+ public void testValidateOptionsThrowsWhenHs2UserIsUsedWithoutHs2Url() throws Exception {
+ expectedException.expect(SqoopOptions.InvalidOptionsException.class);
+ expectedException.expectMessage("The hs2-user option cannot be used without the hs2-url option.");
+
+ when(sqoopOptions.getHs2User()).thenReturn(TEST_HS2_USER);
+
+ sqoopTool.validateOptions(sqoopOptions);
+ }
+
+ @Test
+ public void testValidateOptionsThrowsWhenHs2KeytabIsUsedWithoutHs2User() throws Exception {
+ expectedException.expect(SqoopOptions.InvalidOptionsException.class);
+ expectedException.expectMessage("The hs2-keytab option cannot be used without the hs2-user option.");
+
+ when(sqoopOptions.getHs2Keytab()).thenReturn(TEST_HS2_KEYTAB);
+
+ sqoopTool.validateOptions(sqoopOptions);
+ }
+
+ @Test
+ public void testValidateOptionsSucceedsWhenHs2UrlIsUsedWithHiveImport() throws Exception {
+ when(sqoopOptions.doHiveImport()).thenReturn(true);
+ when(sqoopOptions.getHs2Url()).thenReturn(TEST_HS2_URL);
+
+ sqoopTool.validateOptions(sqoopOptions);
+ }
+
+ @Test
+ public void testValidateOptionsSucceedsWhenHs2UrlIsUsedWithHiveImportAndHs2UserButWithoutHs2Keytab() throws Exception {
+ when(sqoopOptions.doHiveImport()).thenReturn(true);
+ when(sqoopOptions.getHs2Url()).thenReturn(TEST_HS2_URL);
+ when(sqoopOptions.getHs2User()).thenReturn(TEST_HS2_URL);
+
+ sqoopTool.validateOptions(sqoopOptions);
+ }
+
+ @Test
+ public void testValidateOptionsFailsWhenHs2UrlIsUsedWithParquetFormat() throws Exception {
+ expectedException.expect(SqoopOptions.InvalidOptionsException.class);
+ expectedException.expectMessage("The hs2-url option cannot be used with the as-parquetfile option.");
+
+ when(sqoopOptions.doHiveImport()).thenReturn(true);
+ when(sqoopOptions.getHs2Url()).thenReturn(TEST_HS2_URL);
+ when(sqoopOptions.getFileLayout()).thenReturn(ParquetFile);
+
+ sqoopTool.validateOptions(sqoopOptions);
+ }
+
+}
diff --git a/src/test/org/apache/sqoop/tool/TestImportTool.java b/src/test/org/apache/sqoop/tool/TestImportTool.java
index 1c0cf4d..3bdc5c6 100644
--- a/src/test/org/apache/sqoop/tool/TestImportTool.java
+++ b/src/test/org/apache/sqoop/tool/TestImportTool.java
@@ -23,7 +23,6 @@
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -33,10 +32,10 @@
import java.sql.Connection;
import org.apache.sqoop.SqoopOptions.InvalidOptionsException;
-import org.apache.sqoop.hive.HiveImport;
import org.apache.avro.Schema;
import org.apache.sqoop.SqoopOptions;
import org.apache.sqoop.avro.AvroSchemaMismatchException;
+import org.apache.sqoop.hive.HiveClientFactory;
import org.apache.sqoop.util.ExpectedLogMessage;
import org.junit.Assert;
import org.junit.Rule;
@@ -75,7 +74,7 @@
final String actualSchemaString = "actualSchema";
final String errorMessage = "Import failed";
- ImportTool importTool = spy(new ImportTool("import", mock(CodeGenTool.class), false));
+ ImportTool importTool = spy(new ImportTool("import", mock(CodeGenTool.class), false, mock(HiveClientFactory.class)));
doReturn(true).when(importTool).init(any(SqoopOptions.class));
@@ -85,7 +84,7 @@
when(actualSchema.toString()).thenReturn(actualSchemaString);
AvroSchemaMismatchException expectedException = new AvroSchemaMismatchException(errorMessage, writtenWithSchema, actualSchema);
- doThrow(expectedException).when(importTool).importTable(any(SqoopOptions.class), anyString(), any(HiveImport.class));
+ doThrow(expectedException).when(importTool).importTable(any(SqoopOptions.class));
SqoopOptions sqoopOptions = mock(SqoopOptions.class);
when(sqoopOptions.doHiveImport()).thenReturn(true);
