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

SQOOP-2396: Sqoop2: Race condition in purge/update threads on Server shutdown

Browse Source 
(Dian Fu via Jarek Jarcec Cecho)
This commit is contained in:
Jarek Jarcec Cecho 2015-11-18 08:06:49 -08:00
parent 4f0e28625a
commit 190b78fcb8
1 changed files with 43 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
core/src/main/java/org/apache/sqoop/driver/JobManager.java
View File

@ -139,6 +139,16 @@ public static void setInstance(JobManager newInstance) {
*/
private UpdateThread updateThread = null;
/**
* Lock for purge thread.
*/
private Object purgeThreadLock = new Object();
/**
* Lock for update thread.
*/
private Object updateThreadLock = new Object();
/**
* Synchronization variable between threads.
*/
@ -196,20 +206,24 @@ public synchronized void destroy() {
running = false;
try {
purgeThread.interrupt();
purgeThread.join();
} catch (InterruptedException e) {
// TODO(jarcec): Do I want to wait until it actually finish here?
LOG.error("Interrupted joining purgeThread");
synchronized(purgeThreadLock) {
try {
purgeThread.interrupt();
purgeThread.join();
} catch (InterruptedException e) {
// TODO(jarcec): Do I want to wait until it actually finish here?
LOG.error("Interrupted joining purgeThread");
}
}
try {
updateThread.interrupt();
updateThread.join();
} catch (InterruptedException e) {
// TODO(jarcec): Do I want to wait until it actually finish here?
LOG.error("Interrupted joining updateThread");
synchronized(updateThreadLock) {
try {
updateThread.interrupt();
updateThread.join();
} catch (InterruptedException e) {
// TODO(jarcec): Do I want to wait until it actually finish here?
LOG.error("Interrupted joining updateThread");
}
}
if (submissionEngine != null) {
@ -763,11 +777,15 @@ public void run() {
try {
LOG.info("Purging old submissions");
Date threshold = new Date((new Date()).getTime() - purgeThreshold);
RepositoryManager.getInstance().getRepository()
.purgeSubmissions(threshold);
synchronized(purgeThreadLock) {
RepositoryManager.getInstance().getRepository()
.purgeSubmissions(threshold);
}
Thread.sleep(purgeSleep);
} catch (InterruptedException e) {
LOG.debug("Purge thread interrupted", e);
} catch (SqoopException ex) {
LOG.error("Purge thread encountered exception", ex);
}
}
@ -787,18 +805,21 @@ public void run() {
try {
LOG.debug("Updating running submissions");
// Let's get all running submissions from repository to check them out
List<MSubmission> unfinishedSubmissions =
RepositoryManager.getInstance().getRepository()
.findUnfinishedSubmissions();
synchronized(updateThreadLock) {
// Let's get all running submissions from repository to check them out
List<MSubmission> unfinishedSubmissions =
RepositoryManager.getInstance().getRepository()
.findUnfinishedSubmissions();
for (MSubmission submission : unfinishedSubmissions) {
updateSubmission(submission);
for (MSubmission submission : unfinishedSubmissions) {
updateSubmission(submission);
}
}
Thread.sleep(updateSleep);
} catch (InterruptedException e) {
LOG.debug("Purge thread interrupted", e);
LOG.debug("Update thread interrupted", e);
} catch (SqoopException ex) {
LOG.error("Update thread encountered exception", ex);
}
}