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

HDFS-13731. ReencryptionUpdater fails with ConcurrentModificationException during processCheckpoints. Contributed by Zsolt Venczel. 

(cherry picked from commit 3e18b957eb)
This commit is contained in:
Zsolt Venczel 2018-08-28 15:11:58 -07:00 committed by Xiao Chen
parent 450ba6790d
commit 5cbb9b1ca9
2 changed files with 30 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ReencryptionHandler.java
View File

@ -713,10 +713,10 @@ public class ReencryptionHandler implements Runnable {
zst = new ZoneSubmissionTracker(); zst = new ZoneSubmissionTracker();
submissions.put(zoneId, zst); submissions.put(zoneId, zst);
} }
}
Future future = batchService.submit(new EDEKReencryptCallable(zoneId, Future future = batchService.submit(new EDEKReencryptCallable(zoneId,
currentBatch, reencryptionHandler)); currentBatch, reencryptionHandler));
zst.addTask(future); zst.addTask(future);
}
LOG.info("Submitted batch (start:{}, size:{}) of zone {} to re-encrypt.", LOG.info("Submitted batch (start:{}, size:{}) of zone {} to re-encrypt.",
currentBatch.getFirstFilePath(), currentBatch.size(), zoneId); currentBatch.getFirstFilePath(), currentBatch.size(), zoneId);
currentBatch = new ReencryptionBatch(reencryptBatchSize); currentBatch = new ReencryptionBatch(reencryptBatchSize);

6
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ReencryptionUpdater.java
View File

@ -383,6 +383,7 @@ public final class ReencryptionUpdater implements Runnable {
final LinkedList<Future> tasks = tracker.getTasks(); final LinkedList<Future> tasks = tracker.getTasks();
final List<XAttr> xAttrs = Lists.newArrayListWithCapacity(1); final List<XAttr> xAttrs = Lists.newArrayListWithCapacity(1);
ListIterator<Future> iter = tasks.listIterator(); ListIterator<Future> iter = tasks.listIterator();
synchronized (handler) {
while (iter.hasNext()) { while (iter.hasNext()) {
Future<ReencryptionTask> curr = iter.next(); Future<ReencryptionTask> curr = iter.next();
if (curr.isCancelled()) { if (curr.isCancelled()) {
@ -403,13 +404,14 @@ public final class ReencryptionUpdater implements Runnable {
xAttrs.clear(); xAttrs.clear();
xAttrs.add(xattr); xAttrs.add(xattr);
} catch (IOException ie) { } catch (IOException ie) {
LOG.warn("Failed to update re-encrypted progress to xattr for zone {}", LOG.warn("Failed to update re-encrypted progress to xattr" +
zonePath, ie); " for zone {}", zonePath, ie);
++task.numFailures; ++task.numFailures;
} }
++tracker.numCheckpointed; ++tracker.numCheckpointed;
iter.remove(); iter.remove();
} }
}
if (tracker.isCompleted()) { if (tracker.isCompleted()) {
LOG.debug("Removed re-encryption tracker for zone {} because it completed" LOG.debug("Removed re-encryption tracker for zone {} because it completed"
+ " with {} tasks.", zonePath, tracker.numCheckpointed); + " with {} tasks.", zonePath, tracker.numCheckpointed);