diff --git a/build.xml b/build.xml
index 1cee3808..445e16bc 100644
--- a/build.xml
+++ b/build.xml
@@ -73,8 +73,18 @@
+
+
+
+
+
+
+
+
+
+
-
+
diff --git a/ivy.xml b/ivy.xml
index 49904de4..effecb3a 100644
--- a/ivy.xml
+++ b/ivy.xml
@@ -40,11 +40,13 @@ under the License.
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/java/org/apache/sqoop/mapreduce/AutoProgressMapper.java b/src/java/org/apache/sqoop/mapreduce/AutoProgressMapper.java
index 4b61321e..95086417 100644
--- a/src/java/org/apache/sqoop/mapreduce/AutoProgressMapper.java
+++ b/src/java/org/apache/sqoop/mapreduce/AutoProgressMapper.java
@@ -21,7 +21,6 @@
import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
/**
* Identity mapper that continuously reports progress via a background thread.
@@ -32,25 +31,6 @@ public class AutoProgressMapper
public static final Log LOG = LogFactory.getLog(
AutoProgressMapper.class.getName());
- /**
- * Total number of millis for which progress will be reported by the
- * auto-progress thread. If this is zero, then the auto-progress thread will
- * never voluntarily exit.
- */
- private int maxProgressPeriod;
-
- /**
- * Number of milliseconds to sleep for between loop iterations. Must be less
- * than report interval.
- */
- private int sleepInterval;
-
- /**
- * Number of milliseconds between calls to Reporter.progress().
- * Should be a multiple of the sleepInterval.
- */
- private int reportInterval;
-
public static final String MAX_PROGRESS_PERIOD_KEY =
"sqoop.mapred.auto.progress.max";
public static final String SLEEP_INTERVAL_KEY =
@@ -67,112 +47,14 @@ public class AutoProgressMapper
// Disable max progress, by default.
public static final int DEFAULT_MAX_PROGRESS = 0;
- private class ProgressThread extends Thread {
-
- private volatile boolean keepGoing; // While this is true, thread runs.
-
- private Context context;
- private long startTimeMillis;
- private long lastReportMillis;
-
- public ProgressThread(final Context ctxt) {
- this.context = ctxt;
- this.keepGoing = true;
- }
-
- public void signalShutdown() {
- this.keepGoing = false; // volatile update.
- this.interrupt();
- }
-
- public void run() {
- this.lastReportMillis = System.currentTimeMillis();
- this.startTimeMillis = this.lastReportMillis;
-
- final long MAX_PROGRESS = AutoProgressMapper.this.maxProgressPeriod;
- final long REPORT_INTERVAL = AutoProgressMapper.this.reportInterval;
- final long SLEEP_INTERVAL = AutoProgressMapper.this.sleepInterval;
-
- // In a loop:
- // * Check that we haven't run for too long (maxProgressPeriod).
- // * If it's been a report interval since we last made progress,
- // make more.
- // * Sleep for a bit.
- // * If the parent thread has signaled for exit, do so.
- while (this.keepGoing) {
- long curTimeMillis = System.currentTimeMillis();
-
- if (MAX_PROGRESS != 0
- && curTimeMillis - this.startTimeMillis > MAX_PROGRESS) {
- this.keepGoing = false;
- LOG.info("Auto-progress thread exiting after " + MAX_PROGRESS
- + " ms.");
- break;
- }
-
- if (curTimeMillis - this.lastReportMillis > REPORT_INTERVAL) {
- // It's been a full report interval -- claim progress.
- LOG.debug("Auto-progress thread reporting progress");
- this.context.progress();
- this.lastReportMillis = curTimeMillis;
- }
-
- // Unless we got an interrupt while we were working,
- // sleep a bit before doing more work.
- if (!Thread.interrupted()) {
- try {
- Thread.sleep(SLEEP_INTERVAL);
- } catch (InterruptedException ie) {
- // we were notified on something; not necessarily an error.
- }
- }
- }
-
- LOG.info("Auto-progress thread is finished. keepGoing=" + this.keepGoing);
- }
- }
-
- /**
- * Set configuration parameters for the auto-progress thread.
- */
- private void configureAutoProgress(Configuration job) {
- this.maxProgressPeriod = job.getInt(MAX_PROGRESS_PERIOD_KEY,
- DEFAULT_MAX_PROGRESS);
- this.sleepInterval = job.getInt(SLEEP_INTERVAL_KEY,
- DEFAULT_SLEEP_INTERVAL);
- this.reportInterval = job.getInt(REPORT_INTERVAL_KEY,
- DEFAULT_REPORT_INTERVAL);
-
- if (this.reportInterval < 1) {
- LOG.warn("Invalid " + REPORT_INTERVAL_KEY + "; setting to "
- + DEFAULT_REPORT_INTERVAL);
- this.reportInterval = DEFAULT_REPORT_INTERVAL;
- }
-
- if (this.sleepInterval > this.reportInterval || this.sleepInterval < 1) {
- LOG.warn("Invalid " + SLEEP_INTERVAL_KEY + "; setting to "
- + DEFAULT_SLEEP_INTERVAL);
- this.sleepInterval = DEFAULT_SLEEP_INTERVAL;
- }
-
- if (this.maxProgressPeriod < 0) {
- LOG.warn("Invalid " + MAX_PROGRESS_PERIOD_KEY + "; setting to "
- + DEFAULT_MAX_PROGRESS);
- this.maxProgressPeriod = DEFAULT_MAX_PROGRESS;
- }
- }
-
-
// map() method intentionally omitted; Mapper.map() is the identity mapper.
-
/**
* Run the mapping process for this task, wrapped in an auto-progress system.
*/
@Override
public void run(Context context) throws IOException, InterruptedException {
- configureAutoProgress(context.getConfiguration());
- ProgressThread thread = this.new ProgressThread(context);
+ ProgressThread thread = new ProgressThread(context, LOG);
try {
thread.setDaemon(true);
@@ -191,7 +73,7 @@ public void run(Context context) throws IOException, InterruptedException {
LOG.debug("Progress thread shutdown detected.");
} catch (InterruptedException ie) {
LOG.warn("Interrupted when waiting on auto-progress thread: "
- + ie.toString());
+ + ie.toString(), ie);
}
}
}