Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-16646 Enhance LoadIncrementalHFiles API to accept store file paths as input - addendum adheres to original cleanup logic

This commit is contained in:
tedyu 2016-09-20 09:38:18 -07:00
parent 348eb2834a
commit 08d9a2b662
1 changed files with 84 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java
View File

@ -333,6 +333,26 @@ public class LoadIncrementalHFiles extends Configured implements Tool {
doBulkLoad(hfofDir, admin, table, regionLocator, false); doBulkLoad(hfofDir, admin, table, regionLocator, false);
} }
void cleanup(Admin admin, Deque<LoadQueueItem> queue, ExecutorService pool,
SecureBulkLoadClient secureClient) throws IOException {
fsDelegationToken.releaseDelegationToken();
if (bulkToken != null && secureClient != null) {
secureClient.cleanupBulkLoad(admin.getConnection(), bulkToken);
}
if (pool != null) {
pool.shutdown();
}
if (!queue.isEmpty()) {
StringBuilder err = new StringBuilder();
err.append("-------------------------------------------------\n");
err.append("Bulk load aborted with some files not yet loaded:\n");
err.append("-------------------------------------------------\n");
for (LoadQueueItem q : queue) {
err.append(" ").append(q.hfilePath).append('\n');
}
LOG.error(err);
}
}
/** /**
* Perform a bulk load of the given directory into the given * Perform a bulk load of the given directory into the given
* pre-existing table. This method is not threadsafe. * pre-existing table. This method is not threadsafe.
@ -352,12 +372,20 @@ public class LoadIncrementalHFiles extends Configured implements Tool {
// LQI queue does not need to be threadsafe -- all operations on this queue // LQI queue does not need to be threadsafe -- all operations on this queue
// happen in this thread // happen in this thread
Deque<LoadQueueItem> queue = new LinkedList<>(); Deque<LoadQueueItem> queue = new LinkedList<>();
ExecutorService pool = null;
SecureBulkLoadClient secureClient = null;
try {
prepareHFileQueue(map, table, queue, silence); prepareHFileQueue(map, table, queue, silence);
if (queue.isEmpty()) { if (queue.isEmpty()) {
LOG.warn("Bulk load operation did not get any files to load"); LOG.warn("Bulk load operation did not get any files to load");
return; return;
} }
performBulkLoad(admin, table, regionLocator, queue); pool = createExecutorService();
secureClient = new SecureBulkLoadClient(table.getConfiguration(), table);
performBulkLoad(admin, table, regionLocator, queue, pool, secureClient);
} finally {
cleanup(admin, queue, pool, secureClient);
}
} }
/** /**
@ -392,6 +420,9 @@ public class LoadIncrementalHFiles extends Configured implements Tool {
// LQI queue does not need to be threadsafe -- all operations on this queue // LQI queue does not need to be threadsafe -- all operations on this queue
// happen in this thread // happen in this thread
Deque<LoadQueueItem> queue = new LinkedList<>(); Deque<LoadQueueItem> queue = new LinkedList<>();
ExecutorService pool = null;
SecureBulkLoadClient secureClient = null;
try {
prepareHFileQueue(hfofDir, table, queue, validateHFile, silence); prepareHFileQueue(hfofDir, table, queue, validateHFile, silence);
if (queue.isEmpty()) { if (queue.isEmpty()) {
@ -400,16 +431,17 @@ public class LoadIncrementalHFiles extends Configured implements Tool {
"subdirectories that correspond to column family names?"); "subdirectories that correspond to column family names?");
return; return;
} }
performBulkLoad(admin, table, regionLocator, queue); pool = createExecutorService();
secureClient = new SecureBulkLoadClient(table.getConfiguration(), table);
performBulkLoad(admin, table, regionLocator, queue, pool, secureClient);
} finally {
cleanup(admin, queue, pool, secureClient);
}
} }
void performBulkLoad(final Admin admin, Table table, RegionLocator regionLocator, void performBulkLoad(final Admin admin, Table table, RegionLocator regionLocator,
Deque<LoadQueueItem> queue) throws IOException { Deque<LoadQueueItem> queue, ExecutorService pool,
ExecutorService pool = createExecutorService(); SecureBulkLoadClient secureClient) throws IOException {
SecureBulkLoadClient secureClient = new SecureBulkLoadClient(table.getConfiguration(), table);
try {
int count = 0; int count = 0;
if(isSecureBulkLoadEndpointAvailable()) { if(isSecureBulkLoadEndpointAvailable()) {
@ -457,24 +489,6 @@ public class LoadIncrementalHFiles extends Configured implements Tool {
// that we can atomically pull out the groups we want to retry. // that we can atomically pull out the groups we want to retry.
} }
} finally {
fsDelegationToken.releaseDelegationToken();
if(bulkToken != null) {
secureClient.cleanupBulkLoad(admin.getConnection(), bulkToken);
}
pool.shutdown();
if (!queue.isEmpty()) {
StringBuilder err = new StringBuilder();
err.append("-------------------------------------------------\n");
err.append("Bulk load aborted with some files not yet loaded:\n");
err.append("-------------------------------------------------\n");
for (LoadQueueItem q : queue) {
err.append(" ").append(q.hfilePath).append('\n');
}
LOG.error(err);
}
}
if (!queue.isEmpty()) { if (!queue.isEmpty()) {
throw new RuntimeException("Bulk load aborted with some files not yet loaded." throw new RuntimeException("Bulk load aborted with some files not yet loaded."
+ "Please check log for more details."); + "Please check log for more details.");