github/sqoop
5
0
Fork 0
mirror of https://github.com/apache/sqoop.git synced 2025-05-10 12:59:15 +08:00
Code Issues Projects Releases Wiki Activity

SQOOP-813: LoaderExecutor might get into deadlock when exception is raised outside Loader itself

Browse Source 
(Jarek Jarcec Cecho via Cheolsoo Park)
This commit is contained in:
Cheolsoo Park 2013-01-22 16:06:29 -08:00
parent 612060139a
commit 5be8eb6808
1 changed files with 32 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopOutputFormatLoadExecutor.java
View File

@ -175,44 +175,39 @@ private class ConsumerThread implements Runnable {
@Override @Override
public void run() { public void run() {
DataReader reader = new OutputFormatDataReader(); LOG.info("SqoopOutputFormatLoadExecutor consumer thread is starting");
Configuration conf = null;
if (!isTest) {
conf = context.getConfiguration();
loaderName = conf.get(JobConstants.JOB_ETL_LOADER);
}
Loader loader = (Loader) ClassUtils.instantiate(loaderName);
// Objects that should be pass to the Executor execution
PrefixContext subContext = null;
Object configConnection = null;
Object configJob = null;
if (!isTest) {
switch (ConfigurationUtils.getJobType(conf)) {
case EXPORT:
subContext = new PrefixContext(conf, JobConstants.PREFIX_CONNECTOR_CONTEXT);
configConnection = ConfigurationUtils.getConnectorConnection(conf);
configJob = ConfigurationUtils.getConnectorJob(conf);
break;
case IMPORT:
subContext = new PrefixContext(conf, "");
configConnection = ConfigurationUtils.getFrameworkConnection(conf);
configJob = ConfigurationUtils.getFrameworkJob(conf);
break;
default:
readerFinished = true;
// Release so that the writer can tell the framework something went
// wrong.
free.release();
throw new SqoopException(MapreduceExecutionError.MAPRED_EXEC_0023);
}
}
try { try {
DataReader reader = new OutputFormatDataReader();
Configuration conf = null;
if (!isTest) {
conf = context.getConfiguration();
loaderName = conf.get(JobConstants.JOB_ETL_LOADER);
}
Loader loader = (Loader) ClassUtils.instantiate(loaderName);
// Objects that should be pass to the Executor execution
PrefixContext subContext = null;
Object configConnection = null;
Object configJob = null;
if (!isTest) {
switch (ConfigurationUtils.getJobType(conf)) {
case EXPORT:
subContext = new PrefixContext(conf, JobConstants.PREFIX_CONNECTOR_CONTEXT);
configConnection = ConfigurationUtils.getConnectorConnection(conf);
configJob = ConfigurationUtils.getConnectorJob(conf);
break;
case IMPORT:
subContext = new PrefixContext(conf, "");
configConnection = ConfigurationUtils.getFrameworkConnection(conf);
configJob = ConfigurationUtils.getFrameworkJob(conf);
break;
default:
throw new SqoopException(MapreduceExecutionError.MAPRED_EXEC_0023);
}
}
LOG.info("Running loader class " + loaderName); LOG.info("Running loader class " + loaderName);
loader.load(subContext, configConnection, configJob, reader); loader.load(subContext, configConnection, configJob, reader);
LOG.info("Loader has finished"); LOG.info("Loader has finished");