You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by dr...@apache.org on 2016/08/04 12:41:50 UTC
[01/30] hadoop git commit: Revert "HDFS-10301. Interleaving
processing of storages from repeated block reports causes false zombie
storage detection, removes valid blocks. Contributed by Vinitha Gankidi."
Repository: hadoop
Updated Branches:
refs/heads/HADOOP-12756 9f473cf90 -> 70c278115
Revert "HDFS-10301. Interleaving processing of storages from repeated block reports causes false zombie storage detection, removes valid blocks. Contributed by Vinitha Gankidi."
This reverts commit 85a20508bd04851d47c24b7562ec2927d5403446.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/c4463f2e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/c4463f2e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/c4463f2e
Branch: refs/heads/HADOOP-12756
Commit: c4463f2ef20d2cb634a1249246f83c451975f3dc
Parents: 9f473cf
Author: Konstantin V Shvachko <sh...@apache.org>
Authored: Mon Aug 1 22:34:40 2016 -0700
Committer: Konstantin V Shvachko <sh...@apache.org>
Committed: Mon Aug 1 22:34:40 2016 -0700
----------------------------------------------------------------------
.../hadoop/hdfs/protocol/BlockListAsLongs.java | 45 -----------
.../server/blockmanagement/BlockManager.java | 52 ++++++-------
.../BlockReportLeaseManager.java | 4 +-
.../blockmanagement/DatanodeDescriptor.java | 29 ++++++-
.../blockmanagement/DatanodeStorageInfo.java | 11 +++
.../hdfs/server/datanode/BPServiceActor.java | 35 ++-------
.../hdfs/server/namenode/NameNodeRpcServer.java | 40 ++++------
.../blockmanagement/TestBlockManager.java | 19 +++--
.../TestNameNodePrunesMissingStorages.java | 80 ++------------------
...TestDnRespectsBlockReportSplitThreshold.java | 33 +-------
.../TestNNHandlesBlockReportPerStorage.java | 34 ++-------
11 files changed, 108 insertions(+), 274 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
index 26340a9..26c7ffb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
@@ -63,34 +63,6 @@ public abstract class BlockListAsLongs implements Iterable<BlockReportReplica> {
public Iterator<BlockReportReplica> iterator() {
return Collections.emptyIterator();
}
- @Override
- public boolean isStorageReport() {
- return false;
- }
- };
-
- // STORAGE_REPORT is used to report all storages in the DN
- public static final BlockListAsLongs STORAGE_REPORT = new BlockListAsLongs() {
- @Override
- public int getNumberOfBlocks() {
- return -1;
- }
- @Override
- public ByteString getBlocksBuffer() {
- return ByteString.EMPTY;
- }
- @Override
- public long[] getBlockListAsLongs() {
- return EMPTY_LONGS;
- }
- @Override
- public Iterator<BlockReportReplica> iterator() {
- return Collections.emptyIterator();
- }
- @Override
- public boolean isStorageReport() {
- return true;
- }
};
/**
@@ -281,13 +253,6 @@ public abstract class BlockListAsLongs implements Iterable<BlockReportReplica> {
abstract public long[] getBlockListAsLongs();
/**
- * Return true for STORAGE_REPORT BlocksListsAsLongs.
- * Otherwise return false.
- * @return boolean
- */
- abstract public boolean isStorageReport();
-
- /**
* Returns a singleton iterator over blocks in the block report. Do not
* add the returned blocks to a collection.
* @return Iterator
@@ -427,11 +392,6 @@ public abstract class BlockListAsLongs implements Iterable<BlockReportReplica> {
}
@Override
- public boolean isStorageReport() {
- return false;
- }
-
- @Override
public Iterator<BlockReportReplica> iterator() {
return new Iterator<BlockReportReplica>() {
final BlockReportReplica block = new BlockReportReplica();
@@ -515,11 +475,6 @@ public abstract class BlockListAsLongs implements Iterable<BlockReportReplica> {
}
@Override
- public boolean isStorageReport() {
- return false;
- }
-
- @Override
public Iterator<BlockReportReplica> iterator() {
return new Iterator<BlockReportReplica>() {
private final BlockReportReplica block = new BlockReportReplica();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
index d927b2a..349b018 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
@@ -2138,7 +2138,7 @@ public class BlockManager implements BlockStatsMXBean {
public boolean processReport(final DatanodeID nodeID,
final DatanodeStorage storage,
final BlockListAsLongs newReport,
- BlockReportContext context) throws IOException {
+ BlockReportContext context, boolean lastStorageInRpc) throws IOException {
namesystem.writeLock();
final long startTime = Time.monotonicNow(); //after acquiring write lock
final long endTime;
@@ -2189,14 +2189,30 @@ public class BlockManager implements BlockStatsMXBean {
storageInfo.receivedBlockReport();
if (context != null) {
- if (context.getTotalRpcs() == context.getCurRpc() + 1) {
- long leaseId = this.getBlockReportLeaseManager().removeLease(node);
- BlockManagerFaultInjector.getInstance().
- removeBlockReportLease(node, leaseId);
+ storageInfo.setLastBlockReportId(context.getReportId());
+ if (lastStorageInRpc) {
+ int rpcsSeen = node.updateBlockReportContext(context);
+ if (rpcsSeen >= context.getTotalRpcs()) {
+ long leaseId = blockReportLeaseManager.removeLease(node);
+ BlockManagerFaultInjector.getInstance().
+ removeBlockReportLease(node, leaseId);
+ List<DatanodeStorageInfo> zombies = node.removeZombieStorages();
+ if (zombies.isEmpty()) {
+ LOG.debug("processReport 0x{}: no zombie storages found.",
+ Long.toHexString(context.getReportId()));
+ } else {
+ for (DatanodeStorageInfo zombie : zombies) {
+ removeZombieReplicas(context, zombie);
+ }
+ }
+ node.clearBlockReportContext();
+ } else {
+ LOG.debug("processReport 0x{}: {} more RPCs remaining in this " +
+ "report.", Long.toHexString(context.getReportId()),
+ (context.getTotalRpcs() - rpcsSeen)
+ );
+ }
}
- LOG.debug("Processing RPC with index {} out of total {} RPCs in "
- + "processReport 0x{}", context.getCurRpc(),
- context.getTotalRpcs(), Long.toHexString(context.getReportId()));
}
} finally {
endTime = Time.monotonicNow();
@@ -2222,26 +2238,6 @@ public class BlockManager implements BlockStatsMXBean {
return !node.hasStaleStorages();
}
- public void removeZombieStorages(DatanodeRegistration nodeReg,
- BlockReportContext context, Set<String> storageIDsInBlockReport)
- throws UnregisteredNodeException {
- namesystem.writeLock();
- DatanodeDescriptor node = this.getDatanodeManager().getDatanode(nodeReg);
- if (node != null) {
- List<DatanodeStorageInfo> zombies =
- node.removeZombieStorages(storageIDsInBlockReport);
- if (zombies.isEmpty()) {
- LOG.debug("processReport 0x{}: no zombie storages found.",
- Long.toHexString(context.getReportId()));
- } else {
- for (DatanodeStorageInfo zombie : zombies) {
- this.removeZombieReplicas(context, zombie);
- }
- }
- }
- namesystem.writeUnlock();
- }
-
private void removeZombieReplicas(BlockReportContext context,
DatanodeStorageInfo zombie) {
LOG.warn("processReport 0x{}: removing zombie storage {}, which no " +
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockReportLeaseManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockReportLeaseManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockReportLeaseManager.java
index 34e0949..7db05c7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockReportLeaseManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockReportLeaseManager.java
@@ -308,10 +308,10 @@ class BlockReportLeaseManager {
return false;
}
if (node.leaseId == 0) {
- LOG.warn("BR lease 0x{} is not found for DN {}, because the DN " +
+ LOG.warn("BR lease 0x{} is not valid for DN {}, because the DN " +
"is not in the pending set.",
Long.toHexString(id), dn.getDatanodeUuid());
- return true;
+ return false;
}
if (pruneIfExpired(monotonicNowMs, node)) {
LOG.warn("BR lease 0x{} is not valid for DN {}, because the lease " +
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
index d807ab6..1646129 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdfs.server.blockmanagement;
import java.util.ArrayList;
+import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -42,6 +43,7 @@ import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
import org.apache.hadoop.hdfs.server.namenode.CachedBlock;
import org.apache.hadoop.hdfs.server.protocol.BlockECReconstructionCommand.BlockECReconstructionInfo;
+import org.apache.hadoop.hdfs.server.protocol.BlockReportContext;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage.State;
import org.apache.hadoop.hdfs.server.protocol.StorageReport;
@@ -152,6 +154,9 @@ public class DatanodeDescriptor extends DatanodeInfo {
public final DecommissioningStatus decommissioningStatus =
new DecommissioningStatus();
+ private long curBlockReportId = 0;
+
+ private BitSet curBlockReportRpcsSeen = null;
private final Map<String, DatanodeStorageInfo> storageMap =
new HashMap<>();
@@ -252,6 +257,20 @@ public class DatanodeDescriptor extends DatanodeInfo {
updateHeartbeatState(StorageReport.EMPTY_ARRAY, 0L, 0L, 0, 0, null);
}
+ public int updateBlockReportContext(BlockReportContext context) {
+ if (curBlockReportId != context.getReportId()) {
+ curBlockReportId = context.getReportId();
+ curBlockReportRpcsSeen = new BitSet(context.getTotalRpcs());
+ }
+ curBlockReportRpcsSeen.set(context.getCurRpc());
+ return curBlockReportRpcsSeen.cardinality();
+ }
+
+ public void clearBlockReportContext() {
+ curBlockReportId = 0;
+ curBlockReportRpcsSeen = null;
+ }
+
public CachedBlocksList getPendingCached() {
return pendingCached;
}
@@ -315,8 +334,7 @@ public class DatanodeDescriptor extends DatanodeInfo {
}
}
- List<DatanodeStorageInfo>
- removeZombieStorages(Set<String> storageIDsInBlockReport) {
+ List<DatanodeStorageInfo> removeZombieStorages() {
List<DatanodeStorageInfo> zombies = null;
synchronized (storageMap) {
Iterator<Map.Entry<String, DatanodeStorageInfo>> iter =
@@ -324,13 +342,18 @@ public class DatanodeDescriptor extends DatanodeInfo {
while (iter.hasNext()) {
Map.Entry<String, DatanodeStorageInfo> entry = iter.next();
DatanodeStorageInfo storageInfo = entry.getValue();
- if (!storageIDsInBlockReport.contains(storageInfo.getStorageID())) {
+ if (storageInfo.getLastBlockReportId() != curBlockReportId) {
+ LOG.info("{} had lastBlockReportId 0x{} but curBlockReportId = 0x{}",
+ storageInfo.getStorageID(),
+ Long.toHexString(storageInfo.getLastBlockReportId()),
+ Long.toHexString(curBlockReportId));
iter.remove();
if (zombies == null) {
zombies = new LinkedList<>();
}
zombies.add(storageInfo);
}
+ storageInfo.setLastBlockReportId(0);
}
}
return zombies == null ? EMPTY_STORAGE_INFO_LIST : zombies;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeStorageInfo.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeStorageInfo.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeStorageInfo.java
index 1b7cd7c..843a8d5 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeStorageInfo.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeStorageInfo.java
@@ -98,6 +98,9 @@ public class DatanodeStorageInfo {
private final FoldedTreeSet<BlockInfo> blocks = new FoldedTreeSet<>();
+ // The ID of the last full block report which updated this storage.
+ private long lastBlockReportId = 0;
+
/** The number of block reports received */
private int blockReportCount = 0;
@@ -162,6 +165,14 @@ public class DatanodeStorageInfo {
this.blockPoolUsed = blockPoolUsed;
}
+ long getLastBlockReportId() {
+ return lastBlockReportId;
+ }
+
+ void setLastBlockReportId(long lastBlockReportId) {
+ this.lastBlockReportId = lastBlockReportId;
+ }
+
State getState() {
return this.state;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java
index f18cf0b..69989fb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java
@@ -367,36 +367,11 @@ class BPServiceActor implements Runnable {
} else {
// Send one block report per message.
for (int r = 0; r < reports.length; r++) {
- StorageBlockReport[] singleReport = {reports[r]};
- DatanodeCommand cmd;
- if (r != reports.length - 1) {
- cmd = bpNamenode.blockReport(bpRegistration, bpos.getBlockPoolId(),
- singleReport, new BlockReportContext(reports.length, r,
- reportId, fullBrLeaseId, true));
- } else {
- StorageBlockReport[] lastSplitReport =
- new StorageBlockReport[perVolumeBlockLists.size()];
- // When block reports are split, the last RPC in the block report
- // has the information about all storages in the block report.
- // See HDFS-10301 for more details. To achieve this, the last RPC
- // has 'n' storage reports, where 'n' is the number of storages in
- // a DN. The actual block replicas are reported only for the
- // last/n-th storage.
- i = 0;
- for(Map.Entry<DatanodeStorage, BlockListAsLongs> kvPair :
- perVolumeBlockLists.entrySet()) {
- lastSplitReport[i++] = new StorageBlockReport(
- kvPair.getKey(), BlockListAsLongs.STORAGE_REPORT);
- if (i == r) {
- lastSplitReport[i] = reports[r];
- break;
- }
- }
- cmd = bpNamenode.blockReport(
- bpRegistration, bpos.getBlockPoolId(), lastSplitReport,
- new BlockReportContext(reports.length, r, reportId,
- fullBrLeaseId, true));
- }
+ StorageBlockReport singleReport[] = { reports[r] };
+ DatanodeCommand cmd = bpNamenode.blockReport(
+ bpRegistration, bpos.getBlockPoolId(), singleReport,
+ new BlockReportContext(reports.length, r, reportId,
+ fullBrLeaseId, true));
numReportsSent++;
numRPCs++;
if (cmd != null) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
index 3f36fcc..6b52949 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
@@ -1435,37 +1435,25 @@ public class NameNodeRpcServer implements NamenodeProtocols {
boolean noStaleStorages = false;
for (int r = 0; r < reports.length; r++) {
final BlockListAsLongs blocks = reports[r].getBlocks();
- if (!blocks.isStorageReport()) {
- //
- // BlockManager.processReport accumulates information of prior calls
- // for the same node and storage, so the value returned by the last
- // call of this loop is the final updated value for noStaleStorage.
- //
- final int index = r;
- noStaleStorages = bm.runBlockOp(new Callable<Boolean>() {
- @Override
- public Boolean call()
- throws IOException {
- return bm.processReport(nodeReg, reports[index].getStorage(),
- blocks, context);
- }
- });
- metrics.incrStorageBlockReportOps();
- }
+ //
+ // BlockManager.processReport accumulates information of prior calls
+ // for the same node and storage, so the value returned by the last
+ // call of this loop is the final updated value for noStaleStorage.
+ //
+ final int index = r;
+ noStaleStorages = bm.runBlockOp(new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws IOException {
+ return bm.processReport(nodeReg, reports[index].getStorage(),
+ blocks, context, (index == reports.length - 1));
+ }
+ });
+ metrics.incrStorageBlockReportOps();
}
BlockManagerFaultInjector.getInstance().
incomingBlockReportRpc(nodeReg, context);
if (nn.getFSImage().isUpgradeFinalized() &&
- context.getTotalRpcs() == context.getCurRpc() + 1) {
- Set<String> storageIDsInBlockReport = new HashSet<>();
- for (StorageBlockReport report : reports) {
- storageIDsInBlockReport.add(report.getStorage().getStorageID());
- }
- bm.removeZombieStorages(nodeReg, context, storageIDsInBlockReport);
- }
-
- if (nn.getFSImage().isUpgradeFinalized() &&
!namesystem.isRollingUpgrade() &&
!nn.isStandbyState() &&
noStaleStorages) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockManager.java
index 8c231d1..394fae9 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockManager.java
@@ -713,12 +713,12 @@ public class TestBlockManager {
reset(node);
bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
- BlockListAsLongs.EMPTY, null);
+ BlockListAsLongs.EMPTY, null, false);
assertEquals(1, ds.getBlockReportCount());
// send block report again, should NOT be processed
reset(node);
bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
- BlockListAsLongs.EMPTY, null);
+ BlockListAsLongs.EMPTY, null, false);
assertEquals(1, ds.getBlockReportCount());
// re-register as if node restarted, should update existing node
@@ -729,7 +729,7 @@ public class TestBlockManager {
// send block report, should be processed after restart
reset(node);
bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
- BlockListAsLongs.EMPTY, null);
+ BlockListAsLongs.EMPTY, null, false);
// Reinitialize as registration with empty storage list pruned
// node.storageMap.
ds = node.getStorageInfos()[0];
@@ -758,7 +758,7 @@ public class TestBlockManager {
reset(node);
doReturn(1).when(node).numBlocks();
bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
- BlockListAsLongs.EMPTY, null);
+ BlockListAsLongs.EMPTY, null, false);
assertEquals(1, ds.getBlockReportCount());
}
@@ -832,7 +832,7 @@ public class TestBlockManager {
assertEquals(0, ds.getBlockReportCount());
bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
builder.build(),
- new BlockReportContext(1, 0, System.nanoTime(), 0, true));
+ new BlockReportContext(1, 0, System.nanoTime(), 0, true), false);
assertEquals(1, ds.getBlockReportCount());
// verify the storage info is correct
@@ -871,7 +871,8 @@ public class TestBlockManager {
assertEquals(0, ds.getBlockReportCount());
bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
generateReport(blocks),
- new BlockReportContext(1, 0, System.nanoTime(), 0, false));
+ new BlockReportContext(1, 0, System.nanoTime(), 0, false),
+ false);
assertEquals(1, ds.getBlockReportCount());
// verify the storage info is correct
for (BlockInfo block : blocks) {
@@ -881,7 +882,8 @@ public class TestBlockManager {
// Send unsorted report
bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
generateReport(blocks),
- new BlockReportContext(1, 0, System.nanoTime(), 0, false));
+ new BlockReportContext(1, 0, System.nanoTime(), 0, false),
+ false);
assertEquals(2, ds.getBlockReportCount());
// verify the storage info is correct
for (BlockInfo block : blocks) {
@@ -892,7 +894,8 @@ public class TestBlockManager {
Collections.sort(blocks);
bm.processReport(node, new DatanodeStorage(ds.getStorageID()),
generateReport(blocks),
- new BlockReportContext(1, 0, System.nanoTime(), 0, true));
+ new BlockReportContext(1, 0, System.nanoTime(), 0, true),
+ false);
assertEquals(3, ds.getBlockReportCount());
// verify the storage info is correct
for (BlockInfo block : blocks) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestNameNodePrunesMissingStorages.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestNameNodePrunesMissingStorages.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestNameNodePrunesMissingStorages.java
index be38afe..b11b48a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestNameNodePrunesMissingStorages.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestNameNodePrunesMissingStorages.java
@@ -19,40 +19,34 @@
package org.apache.hadoop.hdfs.server.blockmanagement;
import com.google.common.base.Supplier;
-import java.util.ArrayList;
-import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSTestUtil;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
-import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
+import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi.FsVolumeReferences;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
-import org.apache.hadoop.hdfs.server.protocol.BlockReportContext;
-import org.apache.hadoop.hdfs.server.protocol.DatanodeCommand;
import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
-import org.apache.hadoop.hdfs.server.protocol.StorageBlockReport;
import org.apache.hadoop.hdfs.server.protocol.StorageReport;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.log4j.Level;
import org.junit.Assert;
import org.junit.Test;
+import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
@@ -61,6 +55,8 @@ import java.io.FileReader;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.Writer;
+import java.util.Arrays;
+import java.util.EnumSet;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
@@ -372,68 +368,4 @@ public class TestNameNodePrunesMissingStorages {
cluster.shutdown();
}
}
-
- @Test(timeout=300000)
- public void testInterleavedFullBlockReports() throws Exception {
- Configuration conf = new HdfsConfiguration();
- conf.setLong(DFSConfigKeys.DFS_BLOCKREPORT_INTERVAL_MSEC_KEY,
- 36000000L);
- int numStoragesPerDatanode = 6;
- final MiniDFSCluster cluster = new MiniDFSCluster
- .Builder(conf).numDataNodes(1)
- .storagesPerDatanode(numStoragesPerDatanode)
- .build();
- try {
- LOG.info("waiting for cluster to become active...");
- cluster.waitActive();
- // Get the datanode registration and the block reports
- DataNode dn = cluster.getDataNodes().get(0);
- final String blockPoolId = cluster.getNamesystem().getBlockPoolId();
- LOG.info("Block pool id: " + blockPoolId);
- final DatanodeRegistration dnR = dn.getDNRegistrationForBP(blockPoolId);
- Map<DatanodeStorage, BlockListAsLongs> perVolumeBlockLists =
- dn.getFSDataset().getBlockReports(blockPoolId);
- final StorageBlockReport[] reports =
- new StorageBlockReport[perVolumeBlockLists.size()];
- int reportIndex = 0;
- for(Map.Entry<DatanodeStorage, BlockListAsLongs> kvPair :
- perVolumeBlockLists.entrySet()) {
- DatanodeStorage dnStorage = kvPair.getKey();
- BlockListAsLongs blockList = kvPair.getValue();
- reports[reportIndex++] =
- new StorageBlockReport(dnStorage, blockList);
- }
- // Get the list of storage ids associated with the datanode
- // before the test
- BlockManager bm =
- cluster.getNameNode().getNamesystem().getBlockManager();
- final DatanodeDescriptor dnDescriptor = bm.getDatanodeManager().
- getDatanode(cluster.getDataNodes().get(0).getDatanodeUuid());
- DatanodeStorageInfo[] storageInfos = dnDescriptor.getStorageInfos();
- // Send the full block report concurrently using
- // numThreads=numStoragesPerDatanode
- ExecutorService executorService = Executors.
- newFixedThreadPool(numStoragesPerDatanode);
- List<Future<DatanodeCommand>> futureList =
- new ArrayList<>(numStoragesPerDatanode);
- for (int i = 0; i < numStoragesPerDatanode; i++) {
- futureList.add(executorService.submit(new Callable<DatanodeCommand>() {
- @Override
- public DatanodeCommand call() throws IOException {
- return cluster.getNameNodeRpc().blockReport(dnR, blockPoolId,
- reports, new BlockReportContext(1, 0, System.nanoTime(),
- 0L, true));
- }
- }));
- }
- for (Future<DatanodeCommand> future: futureList) {
- future.get();
- }
- executorService.shutdown();
- // Verify that the storages match before and after the test
- Assert.assertArrayEquals(storageInfos, dnDescriptor.getStorageInfos());
- } finally {
- cluster.shutdown();
- }
- }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDnRespectsBlockReportSplitThreshold.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDnRespectsBlockReportSplitThreshold.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDnRespectsBlockReportSplitThreshold.java
index f41c546..bf0e3c1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDnRespectsBlockReportSplitThreshold.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDnRespectsBlockReportSplitThreshold.java
@@ -41,7 +41,6 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import static org.hamcrest.core.Is.is;
-import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.*;
import static org.mockito.Mockito.times;
@@ -89,34 +88,6 @@ public class TestDnRespectsBlockReportSplitThreshold {
blockCount * BLOCK_SIZE, BLOCK_SIZE, REPL_FACTOR, seed);
}
- private void verifyCapturedArgumentsSplit(
- ArgumentCaptor<StorageBlockReport[]> captor,
- int expectedReportsPerCall,
- int expectedTotalBlockCount) {
- List<StorageBlockReport[]> listOfReports = captor.getAllValues();
- int numBlocksReported = 0;
- int storageIndex = 0;
- int listOfReportsSize = listOfReports.size();
- for (StorageBlockReport[] reports : listOfReports) {
- if (storageIndex < (listOfReportsSize - 1)) {
- assertThat(reports.length, is(expectedReportsPerCall));
- } else {
- assertThat(reports.length, is(listOfReportsSize));
- }
- for (StorageBlockReport report : reports) {
- BlockListAsLongs blockList = report.getBlocks();
- if (!blockList.isStorageReport()) {
- numBlocksReported += blockList.getNumberOfBlocks();
- } else {
- assertEquals(blockList.getNumberOfBlocks(), -1);
- }
- }
- storageIndex++;
- }
-
- assert(numBlocksReported >= expectedTotalBlockCount);
- }
-
private void verifyCapturedArguments(
ArgumentCaptor<StorageBlockReport[]> captor,
int expectedReportsPerCall,
@@ -165,7 +136,7 @@ public class TestDnRespectsBlockReportSplitThreshold {
anyString(),
captor.capture(), Mockito.<BlockReportContext>anyObject());
- verifyCapturedArgumentsSplit(captor, 1, BLOCKS_IN_FILE);
+ verifyCapturedArguments(captor, 1, BLOCKS_IN_FILE);
}
/**
@@ -229,7 +200,7 @@ public class TestDnRespectsBlockReportSplitThreshold {
anyString(),
captor.capture(), Mockito.<BlockReportContext>anyObject());
- verifyCapturedArgumentsSplit(captor, 1, BLOCKS_IN_FILE);
+ verifyCapturedArguments(captor, 1, BLOCKS_IN_FILE);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c4463f2e/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestNNHandlesBlockReportPerStorage.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestNNHandlesBlockReportPerStorage.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestNNHandlesBlockReportPerStorage.java
index 524243b..791ee20 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestNNHandlesBlockReportPerStorage.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestNNHandlesBlockReportPerStorage.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.hdfs.server.datanode;
import java.io.IOException;
-import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
import org.apache.hadoop.hdfs.server.protocol.BlockReportContext;
import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
import org.apache.hadoop.hdfs.server.protocol.StorageBlockReport;
@@ -35,32 +34,13 @@ public class TestNNHandlesBlockReportPerStorage extends BlockReportTestBase {
@Override
protected void sendBlockReports(DatanodeRegistration dnR, String poolId,
StorageBlockReport[] reports) throws IOException {
- for (int r = 0; r < reports.length; r++) {
- LOG.info("Sending block report for storage " +
- reports[r].getStorage().getStorageID());
- StorageBlockReport[] singletonReport = {reports[r]};
- if (r != reports.length - 1) {
- cluster.getNameNodeRpc().blockReport(dnR, poolId, singletonReport,
- new BlockReportContext(reports.length, r, System.nanoTime(),
- 0L, true));
- } else {
- StorageBlockReport[] lastSplitReport =
- new StorageBlockReport[reports.length];
- // When block reports are split, send a dummy storage report for all
- // other storages in the blockreport along with the last storage report
- for (int i = 0; i <= r; i++) {
- if (i == r) {
- lastSplitReport[i] = reports[r];
- break;
- }
- lastSplitReport[i] =
- new StorageBlockReport(reports[i].getStorage(),
- BlockListAsLongs.STORAGE_REPORT);
- }
- cluster.getNameNodeRpc().blockReport(dnR, poolId, lastSplitReport,
- new BlockReportContext(reports.length, r, System.nanoTime(),
- 0L, true));
- }
+ int i = 0;
+ for (StorageBlockReport report : reports) {
+ LOG.info("Sending block report for storage " + report.getStorage().getStorageID());
+ StorageBlockReport[] singletonReport = { report };
+ cluster.getNameNodeRpc().blockReport(dnR, poolId, singletonReport,
+ new BlockReportContext(reports.length, i, System.nanoTime(), 0L, true));
+ i++;
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[07/30] hadoop git commit: YARN-5226. Remove AHS enable check from
LogsCLI#fetchAMContainerLogs. Contributed by Xuan Gong.
Posted by dr...@apache.org.
YARN-5226. Remove AHS enable check from LogsCLI#fetchAMContainerLogs. Contributed by Xuan Gong.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/38183932
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/38183932
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/38183932
Branch: refs/heads/HADOOP-12756
Commit: 3818393297c7b337e380e8111a55f2ac4745cb83
Parents: b3018e7
Author: Junping Du <ju...@apache.org>
Authored: Tue Aug 2 10:41:46 2016 -0700
Committer: Junping Du <ju...@apache.org>
Committed: Tue Aug 2 10:41:46 2016 -0700
----------------------------------------------------------------------
.../apache/hadoop/yarn/client/cli/LogsCLI.java | 102 ++++++++-----------
1 file changed, 41 insertions(+), 61 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/38183932/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/LogsCLI.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/LogsCLI.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/LogsCLI.java
index e2c4552..c49d463 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/LogsCLI.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/LogsCLI.java
@@ -555,7 +555,10 @@ public class LogsCLI extends Configured implements Tool {
new ArrayList<ContainerLogsRequest>();
boolean getAMContainerLists = false;
String appId = request.getAppId().toString();
- String errorMessage = "";
+ StringBuilder errorMessage = new StringBuilder();
+ // We will call RM webservice to get all AppAttempts information.
+ // If we get nothing, we will try to call AHS webservice to get AppAttempts
+ // which includes nodeAddress for the AM Containers.
try {
amContainersList = getAMContainerInfoForRMWebService(conf, appId);
if (amContainersList != null && !amContainersList.isEmpty()) {
@@ -572,21 +575,28 @@ public class LogsCLI extends Configured implements Tool {
}
}
} catch (Exception ex) {
- errorMessage = ex.getMessage();
+ errorMessage.append(ex.getMessage() + "\n");
if (request.isAppFinished()) {
- try {
- amContainersList = getAMContainerInfoForAHSWebService(conf, appId);
- if (amContainersList != null && !amContainersList.isEmpty()) {
- getAMContainerLists = true;
- for (JSONObject amContainer : amContainersList) {
- ContainerLogsRequest amRequest = new ContainerLogsRequest(
- request);
- amRequest.setContainerId(amContainer.getString("amContainerId"));
- requests.add(amRequest);
+ if (!conf.getBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED,
+ YarnConfiguration.DEFAULT_TIMELINE_SERVICE_ENABLED)) {
+ errorMessage.append("Please enable the timeline service "
+ + "and make sure the timeline server is running.");
+ } else {
+ try {
+ amContainersList = getAMContainerInfoForAHSWebService(conf, appId);
+ if (amContainersList != null && !amContainersList.isEmpty()) {
+ getAMContainerLists = true;
+ for (JSONObject amContainer : amContainersList) {
+ ContainerLogsRequest amRequest = new ContainerLogsRequest(
+ request);
+ amRequest.setContainerId(
+ amContainer.getString("amContainerId"));
+ requests.add(amRequest);
+ }
}
+ } catch (Exception e) {
+ errorMessage.append(e.getMessage());
}
- } catch (Exception e) {
- errorMessage = e.getMessage();
}
}
}
@@ -595,6 +605,9 @@ public class LogsCLI extends Configured implements Tool {
System.err.println("Unable to get AM container informations "
+ "for the application:" + appId);
System.err.println(errorMessage);
+ System.err.println("Can not get AMContainers logs for "
+ + "the application:" + appId + " with the appOwner:"
+ + request.getAppOwner());
return -1;
}
@@ -635,19 +648,12 @@ public class LogsCLI extends Configured implements Tool {
if (request.isAppFinished()) {
if (containerId != null && !containerId.isEmpty()) {
- if (nodeId == null || nodeId.isEmpty()) {
- try {
- nodeId =
- getContainerReport(containerId).getAssignedNode().toString();
- request.setNodeId(nodeId);
- } catch (Exception ex) {
- System.err.println(ex);
- nodeId = null;
- }
- }
if (nodeId != null && !nodeId.isEmpty()) {
printContainerLogsForFinishedApplication(request,
logCliHelper);
+ } else {
+ printContainerLogsForFinishedApplicationWithoutNodeId(
+ request, logCliHelper);
}
}
} else {
@@ -825,39 +831,8 @@ public class LogsCLI extends Configured implements Tool {
private int fetchAMContainerLogs(ContainerLogsRequest request,
List<String> amContainersList, LogCLIHelpers logCliHelper)
throws Exception {
-
- // If the application is running, we will call the RM WebService
- // to get the AppAttempts which includes the nodeHttpAddress
- // and containerId for all the AM Containers.
- // After that, we will call NodeManager webService to get the
- // related logs
- if (!request.isAppFinished()) {
- return printAMContainerLogs(getConf(), request, amContainersList,
- logCliHelper);
- } else {
- // If the application is in the final state, we will call RM webservice
- // to get all AppAttempts information first. If we get nothing,
- // we will try to call AHS webservice to get related AppAttempts
- // which includes nodeAddress for the AM Containers.
- // After that, we will use nodeAddress and containerId
- // to get logs from HDFS directly.
- if (getConf().getBoolean(YarnConfiguration.APPLICATION_HISTORY_ENABLED,
- YarnConfiguration.DEFAULT_APPLICATION_HISTORY_ENABLED)) {
- return printAMContainerLogs(getConf(), request, amContainersList,
- logCliHelper);
- } else {
- ApplicationId appId = request.getAppId();
- String appOwner = request.getAppOwner();
- System.err.println("Can not get AMContainers logs for "
- + "the application:" + appId + " with the appOwner:" + appOwner);
- System.err.println("This application:" + appId + " has finished."
- + " Please enable the application-history service or explicitly"
- + " use 'yarn logs -applicationId <appId> "
- + "-containerId <containerId> --nodeAddress <nodeHttpAddress>' "
- + "to get the container logs.");
- return -1;
- }
- }
+ return printAMContainerLogs(getConf(), request, amContainersList,
+ logCliHelper);
}
private int fetchContainerLogs(ContainerLogsRequest request,
@@ -868,13 +843,18 @@ public class LogsCLI extends Configured implements Tool {
String nodeAddress = request.getNodeId();
String appOwner = request.getAppOwner();
boolean isAppFinished = request.isAppFinished();
- // if we provide the node address and the application is in the final
- // state, we could directly get logs from HDFS.
- if (nodeAddress != null && isAppFinished) {
+ // if the application is in the final state,
+ // we could directly get logs from HDFS.
+ if (isAppFinished) {
// if user specified "ALL" as the logFiles param, pass empty list
// to logCliHelper so that it fetches all the logs
- return printContainerLogsForFinishedApplication(
- request, logCliHelper);
+ if (nodeAddress != null && !nodeAddress.isEmpty()) {
+ return printContainerLogsForFinishedApplication(
+ request, logCliHelper);
+ } else {
+ return printContainerLogsForFinishedApplicationWithoutNodeId(
+ request, logCliHelper);
+ }
}
String nodeHttpAddress = null;
String nodeId = null;
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[18/30] hadoop git commit: HDFS-10674. Optimize creating a full path
from an inode. Contributed by Daryn Sharp.
Posted by dr...@apache.org.
HDFS-10674. Optimize creating a full path from an inode. Contributed by Daryn Sharp.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/22ef5286
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/22ef5286
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/22ef5286
Branch: refs/heads/HADOOP-12756
Commit: 22ef5286bc8511ddee9594b7cecc598bf41a850b
Parents: 58db263
Author: Kihwal Lee <ki...@apache.org>
Authored: Wed Aug 3 13:12:43 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Wed Aug 3 13:12:43 2016 -0500
----------------------------------------------------------------------
.../hdfs/server/namenode/FSDirectory.java | 55 +-------------------
.../hdfs/server/namenode/FSNamesystem.java | 2 +-
.../hadoop/hdfs/server/namenode/INode.java | 20 ++++++-
.../hdfs/server/namenode/INodesInPath.java | 4 +-
.../server/namenode/TestSnapshotPathINodes.java | 7 ++-
5 files changed, 29 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/22ef5286/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java
index 4930c1d..334c358 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java
@@ -885,57 +885,6 @@ public class FSDirectory implements Closeable {
return typeSpaceDeltas;
}
- /** Return the name of the path represented by inodes at [0, pos] */
- static String getFullPathName(INode[] inodes, int pos) {
- StringBuilder fullPathName = new StringBuilder();
- if (inodes[0].isRoot()) {
- if (pos == 0) return Path.SEPARATOR;
- } else {
- fullPathName.append(inodes[0].getLocalName());
- }
-
- for (int i=1; i<=pos; i++) {
- fullPathName.append(Path.SEPARATOR_CHAR).append(inodes[i].getLocalName());
- }
- return fullPathName.toString();
- }
-
- /**
- * @return the relative path of an inode from one of its ancestors,
- * represented by an array of inodes.
- */
- private static INode[] getRelativePathINodes(INode inode, INode ancestor) {
- // calculate the depth of this inode from the ancestor
- int depth = 0;
- for (INode i = inode; i != null && !i.equals(ancestor); i = i.getParent()) {
- depth++;
- }
- INode[] inodes = new INode[depth];
-
- // fill up the inodes in the path from this inode to root
- for (int i = 0; i < depth; i++) {
- if (inode == null) {
- NameNode.stateChangeLog.warn("Could not get full path."
- + " Corresponding file might have deleted already.");
- return null;
- }
- inodes[depth-i-1] = inode;
- inode = inode.getParent();
- }
- return inodes;
- }
-
- private static INode[] getFullPathINodes(INode inode) {
- return getRelativePathINodes(inode, null);
- }
-
- /** Return the full path name of the specified inode */
- static String getFullPathName(INode inode) {
- INode[] inodes = getFullPathINodes(inode);
- // inodes can be null only when its called without holding lock
- return inodes == null ? "" : getFullPathName(inodes, inodes.length - 1);
- }
-
/**
* Add the given child to the namespace.
* @param existing the INodesInPath containing all the ancestral INodes
@@ -987,9 +936,7 @@ public class FSDirectory implements Closeable {
try {
q.verifyQuota(deltas);
} catch (QuotaExceededException e) {
- List<INode> inodes = iip.getReadOnlyINodes();
- final String path = getFullPathName(inodes.toArray(new INode[inodes.size()]), i);
- e.setPathName(path);
+ e.setPathName(iip.getPath(i));
throw e;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/22ef5286/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
index 915ae97..fb87d4c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
@@ -5085,7 +5085,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
final INodeFile inode = getBlockCollection(blk);
skip++;
if (inode != null && blockManager.countNodes(blk).liveReplicas() == 0) {
- String src = FSDirectory.getFullPathName(inode);
+ String src = inode.getFullPathName();
if (src.startsWith(path)){
corruptFiles.add(new CorruptFileBlockInfo(src, blk));
count++;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/22ef5286/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INode.java
index eb910d4..38883b6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INode.java
@@ -569,7 +569,25 @@ public abstract class INode implements INodeAttributes, Diff.Element<byte[]> {
public String getFullPathName() {
// Get the full path name of this inode.
- return FSDirectory.getFullPathName(this);
+ if (isRoot()) {
+ return Path.SEPARATOR;
+ }
+ // compute size of needed bytes for the path
+ int idx = 0;
+ for (INode inode = this; inode != null; inode = inode.getParent()) {
+ // add component + delimiter (if not tail component)
+ idx += inode.getLocalNameBytes().length + (inode != this ? 1 : 0);
+ }
+ byte[] path = new byte[idx];
+ for (INode inode = this; inode != null; inode = inode.getParent()) {
+ if (inode != this) {
+ path[--idx] = Path.SEPARATOR_CHAR;
+ }
+ byte[] name = inode.getLocalNameBytes();
+ idx -= name.length;
+ System.arraycopy(name, 0, path, idx, name.length);
+ }
+ return DFSUtil.bytes2String(path);
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/22ef5286/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java
index 1d540b7..a7b772c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java
@@ -347,11 +347,11 @@ public class INodesInPath {
}
public String getParentPath() {
- return getPath(path.length - 1);
+ return getPath(path.length - 2);
}
public String getPath(int pos) {
- return DFSUtil.byteArray2PathString(path, 0, pos);
+ return DFSUtil.byteArray2PathString(path, 0, pos + 1); // it's a length...
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/22ef5286/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java
index 45c65ef..d022903 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java
@@ -155,7 +155,12 @@ public class TestSnapshotPathINodes {
sub1.toString());
assertEquals(nodesInPath.getINode(components.length - 3).getFullPathName(),
dir.toString());
-
+
+ assertEquals(Path.SEPARATOR, nodesInPath.getPath(0));
+ assertEquals(dir.toString(), nodesInPath.getPath(1));
+ assertEquals(sub1.toString(), nodesInPath.getPath(2));
+ assertEquals(file1.toString(), nodesInPath.getPath(3));
+
nodesInPath = INodesInPath.resolve(fsdir.rootDir, components, false);
assertEquals(nodesInPath.length(), components.length);
assertSnapshot(nodesInPath, false, null, -1);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[30/30] hadoop git commit: MAPREDUCE-6730. Use StandardCharsets
instead of String overload in TextOutputFormat. Contributed by Sahil Kang.
Posted by dr...@apache.org.
MAPREDUCE-6730. Use StandardCharsets instead of String overload in TextOutputFormat. Contributed by Sahil Kang.
This closes #114
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/70c27811
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/70c27811
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/70c27811
Branch: refs/heads/HADOOP-12756
Commit: 70c278115249898132490a89a548fd936c09f54b
Parents: 08e3338
Author: Akira Ajisaka <aa...@apache.org>
Authored: Thu Aug 4 20:13:11 2016 +0900
Committer: Akira Ajisaka <aa...@apache.org>
Committed: Thu Aug 4 20:13:11 2016 +0900
----------------------------------------------------------------------
.../apache/hadoop/mapred/TextOutputFormat.java | 24 ++++++--------------
.../mapreduce/lib/output/TextOutputFormat.java | 24 ++++++--------------
2 files changed, 14 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/70c27811/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TextOutputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TextOutputFormat.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TextOutputFormat.java
index ca499e4..bf2f447 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TextOutputFormat.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TextOutputFormat.java
@@ -20,7 +20,7 @@ package org.apache.hadoop.mapred;
import java.io.DataOutputStream;
import java.io.IOException;
-import java.io.UnsupportedEncodingException;
+import java.nio.charset.StandardCharsets;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
@@ -43,26 +43,16 @@ public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
protected static class LineRecordWriter<K, V>
implements RecordWriter<K, V> {
- private static final String utf8 = "UTF-8";
- private static final byte[] newline;
- static {
- try {
- newline = "\n".getBytes(utf8);
- } catch (UnsupportedEncodingException uee) {
- throw new IllegalArgumentException("can't find " + utf8 + " encoding");
- }
- }
+ private static final byte[] NEWLINE =
+ "\n".getBytes(StandardCharsets.UTF_8);
protected DataOutputStream out;
private final byte[] keyValueSeparator;
public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {
this.out = out;
- try {
- this.keyValueSeparator = keyValueSeparator.getBytes(utf8);
- } catch (UnsupportedEncodingException uee) {
- throw new IllegalArgumentException("can't find " + utf8 + " encoding");
- }
+ this.keyValueSeparator =
+ keyValueSeparator.getBytes(StandardCharsets.UTF_8);
}
public LineRecordWriter(DataOutputStream out) {
@@ -80,7 +70,7 @@ public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
Text to = (Text) o;
out.write(to.getBytes(), 0, to.getLength());
} else {
- out.write(o.toString().getBytes(utf8));
+ out.write(o.toString().getBytes(StandardCharsets.UTF_8));
}
}
@@ -101,7 +91,7 @@ public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
if (!nullValue) {
writeObject(value);
}
- out.write(newline);
+ out.write(NEWLINE);
}
public synchronized void close(Reporter reporter) throws IOException {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/70c27811/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/TextOutputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/TextOutputFormat.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/TextOutputFormat.java
index 1522c4c..1c8ea72 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/TextOutputFormat.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/TextOutputFormat.java
@@ -20,7 +20,7 @@ package org.apache.hadoop.mapreduce.lib.output;
import java.io.DataOutputStream;
import java.io.IOException;
-import java.io.UnsupportedEncodingException;
+import java.nio.charset.StandardCharsets;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
@@ -45,26 +45,16 @@ public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
public static String SEPERATOR = "mapreduce.output.textoutputformat.separator";
protected static class LineRecordWriter<K, V>
extends RecordWriter<K, V> {
- private static final String utf8 = "UTF-8";
- private static final byte[] newline;
- static {
- try {
- newline = "\n".getBytes(utf8);
- } catch (UnsupportedEncodingException uee) {
- throw new IllegalArgumentException("can't find " + utf8 + " encoding");
- }
- }
+ private static final byte[] NEWLINE =
+ "\n".getBytes(StandardCharsets.UTF_8);
protected DataOutputStream out;
private final byte[] keyValueSeparator;
public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {
this.out = out;
- try {
- this.keyValueSeparator = keyValueSeparator.getBytes(utf8);
- } catch (UnsupportedEncodingException uee) {
- throw new IllegalArgumentException("can't find " + utf8 + " encoding");
- }
+ this.keyValueSeparator =
+ keyValueSeparator.getBytes(StandardCharsets.UTF_8);
}
public LineRecordWriter(DataOutputStream out) {
@@ -82,7 +72,7 @@ public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
Text to = (Text) o;
out.write(to.getBytes(), 0, to.getLength());
} else {
- out.write(o.toString().getBytes(utf8));
+ out.write(o.toString().getBytes(StandardCharsets.UTF_8));
}
}
@@ -103,7 +93,7 @@ public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
if (!nullValue) {
writeObject(value);
}
- out.write(newline);
+ out.write(NEWLINE);
}
public synchronized
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[20/30] hadoop git commit: YARN-4280. CapacityScheduler reservations
may not prevent indefinite postponement on a busy cluster. Contributed by
Kuhu Shukla
Posted by dr...@apache.org.
YARN-4280. CapacityScheduler reservations may not prevent indefinite postponement on a busy cluster. Contributed by Kuhu Shukla
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/4d92aefd
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/4d92aefd
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/4d92aefd
Branch: refs/heads/HADOOP-12756
Commit: 4d92aefd35d4517d9435d81bafdec0d77905a7a0
Parents: 580a833
Author: Jason Lowe <jl...@apache.org>
Authored: Wed Aug 3 18:53:14 2016 +0000
Committer: Jason Lowe <jl...@apache.org>
Committed: Wed Aug 3 18:53:14 2016 +0000
----------------------------------------------------------------------
.../scheduler/capacity/CSAssignment.java | 31 +-
.../scheduler/capacity/LeafQueue.java | 7 +-
.../scheduler/capacity/ParentQueue.java | 43 ++-
.../allocator/AbstractContainerAllocator.java | 11 +-
.../allocator/IncreaseContainerAllocator.java | 10 +-
.../capacity/TestCapacityScheduler.java | 288 ++++++++++++++++++-
6 files changed, 362 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4d92aefd/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSAssignment.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSAssignment.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSAssignment.java
index 6406efe..7bea9af 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSAssignment.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSAssignment.java
@@ -34,36 +34,47 @@ public class CSAssignment {
public static final CSAssignment NULL_ASSIGNMENT =
new CSAssignment(Resources.createResource(0, 0), NodeType.NODE_LOCAL);
- public static final CSAssignment SKIP_ASSIGNMENT = new CSAssignment(true);
+ public static final CSAssignment SKIP_ASSIGNMENT =
+ new CSAssignment(SkippedType.OTHER);
private Resource resource;
private NodeType type;
private RMContainer excessReservation;
private FiCaSchedulerApp application;
- private final boolean skipped;
+ private SkippedType skipped;
+
+ /**
+ * Reason for the queue to get skipped.
+ */
+ public enum SkippedType {
+ NONE,
+ QUEUE_LIMIT,
+ OTHER
+ }
+
private boolean fulfilledReservation;
private final AssignmentInformation assignmentInformation;
private boolean increaseAllocation;
private List<RMContainer> containersToKill;
public CSAssignment(Resource resource, NodeType type) {
- this(resource, type, null, null, false, false);
+ this(resource, type, null, null, SkippedType.NONE, false);
}
public CSAssignment(FiCaSchedulerApp application,
RMContainer excessReservation) {
this(excessReservation.getContainer().getResource(), NodeType.NODE_LOCAL,
- excessReservation, application, false, false);
+ excessReservation, application, SkippedType.NONE, false);
}
- public CSAssignment(boolean skipped) {
+ public CSAssignment(SkippedType skipped) {
this(Resource.newInstance(0, 0), NodeType.NODE_LOCAL, null, null, skipped,
false);
}
public CSAssignment(Resource resource, NodeType type,
RMContainer excessReservation, FiCaSchedulerApp application,
- boolean skipped, boolean fulfilledReservation) {
+ SkippedType skipped, boolean fulfilledReservation) {
this.resource = resource;
this.type = type;
this.excessReservation = excessReservation;
@@ -105,10 +116,14 @@ public class CSAssignment {
excessReservation = rmContainer;
}
- public boolean getSkipped() {
+ public SkippedType getSkippedType() {
return skipped;
}
-
+
+ public void setSkippedType(SkippedType skippedType) {
+ this.skipped = skippedType;
+ }
+
@Override
public String toString() {
String ret = "resource:" + resource.toString();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4d92aefd/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
index 9aae909..a243e93 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
@@ -60,6 +60,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
+
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedContainerChangeRequest;
@@ -972,8 +973,12 @@ public class LeafQueue extends AbstractCSQueue {
// Done
return assignment;
- } else if (assignment.getSkipped()) {
+ } else if (assignment.getSkippedType()
+ == CSAssignment.SkippedType.OTHER) {
application.updateNodeInfoForAMDiagnostics(node);
+ } else if(assignment.getSkippedType()
+ == CSAssignment.SkippedType.QUEUE_LIMIT) {
+ return assignment;
} else {
// If we don't allocate anything, and it is not skipped by application,
// we will return to respect FIFO of applications
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4d92aefd/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java
index 6fcd6c1..9ae35ee 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java
@@ -37,7 +37,6 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.exceptions.InvalidResourceRequestException;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
-import org.apache.hadoop.yarn.nodelabels.RMNodeLabel;
import org.apache.hadoop.yarn.security.AccessType;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
@@ -474,6 +473,7 @@ public class ParentQueue extends AbstractCSQueue {
" cluster=" + clusterResource);
} else {
+ assignment.setSkippedType(assignedToChild.getSkippedType());
break;
}
@@ -511,14 +511,14 @@ public class ParentQueue extends AbstractCSQueue {
}
private ResourceLimits getResourceLimitsOfChild(CSQueue child,
- Resource clusterResource, ResourceLimits parentLimits,
+ Resource clusterResource, Resource parentLimits,
String nodePartition) {
// Set resource-limit of a given child, child.limit =
// min(my.limit - my.used + child.used, child.max)
// Parent available resource = parent-limit - parent-used-resource
Resource parentMaxAvailableResource = Resources.subtract(
- parentLimits.getLimit(), queueUsage.getUsed(nodePartition));
+ parentLimits, queueUsage.getUsed(nodePartition));
// Deduct killable from used
Resources.addTo(parentMaxAvailableResource,
getTotalKillableResource(nodePartition));
@@ -568,9 +568,9 @@ public class ParentQueue extends AbstractCSQueue {
private synchronized CSAssignment assignContainersToChildQueues(
Resource cluster, FiCaSchedulerNode node, ResourceLimits limits,
SchedulingMode schedulingMode) {
- CSAssignment assignment =
- new CSAssignment(Resources.createResource(0, 0), NodeType.NODE_LOCAL);
-
+ CSAssignment assignment = CSAssignment.NULL_ASSIGNMENT;
+
+ Resource parentLimits = limits.getLimit();
printChildQueues();
// Try to assign to most 'under-served' sub-queue
@@ -584,20 +584,21 @@ public class ParentQueue extends AbstractCSQueue {
// Get ResourceLimits of child queue before assign containers
ResourceLimits childLimits =
- getResourceLimitsOfChild(childQueue, cluster, limits, node.getPartition());
+ getResourceLimitsOfChild(childQueue, cluster, parentLimits,
+ node.getPartition());
- assignment = childQueue.assignContainers(cluster, node,
+ CSAssignment childAssignment = childQueue.assignContainers(cluster, node,
childLimits, schedulingMode);
if(LOG.isDebugEnabled()) {
LOG.debug("Assigned to queue: " + childQueue.getQueuePath() +
- " stats: " + childQueue + " --> " +
- assignment.getResource() + ", " + assignment.getType());
+ " stats: " + childQueue + " --> " +
+ childAssignment.getResource() + ", " + childAssignment.getType());
}
// If we do assign, remove the queue and re-insert in-order to re-sort
if (Resources.greaterThan(
resourceCalculator, cluster,
- assignment.getResource(), Resources.none())) {
+ childAssignment.getResource(), Resources.none())) {
// Only update childQueues when we doing non-partitioned node
// allocation.
if (RMNodeLabelsManager.NO_LABEL.equals(node.getPartition())) {
@@ -610,7 +611,24 @@ public class ParentQueue extends AbstractCSQueue {
printChildQueues();
}
}
+ assignment = childAssignment;
break;
+ } else if (childAssignment.getSkippedType() ==
+ CSAssignment.SkippedType.QUEUE_LIMIT) {
+ if (assignment.getSkippedType() !=
+ CSAssignment.SkippedType.QUEUE_LIMIT) {
+ assignment = childAssignment;
+ }
+ Resource resourceToSubtract = Resources.max(resourceCalculator,
+ cluster, childLimits.getHeadroom(), Resources.none());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Decrease parentLimits " + parentLimits +
+ " for " + this.getQueueName() + " by " +
+ resourceToSubtract + " as childQueue=" +
+ childQueue.getQueueName() + " is blocked");
+ }
+ parentLimits = Resources.subtract(parentLimits,
+ resourceToSubtract);
}
}
@@ -731,7 +749,8 @@ public class ParentQueue extends AbstractCSQueue {
for (CSQueue childQueue : childQueues) {
// Get ResourceLimits of child queue before assign containers
ResourceLimits childLimits = getResourceLimitsOfChild(childQueue,
- clusterResource, resourceLimits, RMNodeLabelsManager.NO_LABEL);
+ clusterResource, resourceLimits.getLimit(),
+ RMNodeLabelsManager.NO_LABEL);
childQueue.updateClusterResource(clusterResource, childLimits);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4d92aefd/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/AbstractContainerAllocator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/AbstractContainerAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/AbstractContainerAllocator.java
index afac235..4d5a7dc 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/AbstractContainerAllocator.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/AbstractContainerAllocator.java
@@ -55,8 +55,10 @@ public abstract class AbstractContainerAllocator {
Resource clusterResource, ContainerAllocation result,
RMContainer rmContainer) {
// Handle skipped
- boolean skipped =
- (result.getAllocationState() == AllocationState.APP_SKIPPED);
+ CSAssignment.SkippedType skipped =
+ (result.getAllocationState() == AllocationState.APP_SKIPPED) ?
+ CSAssignment.SkippedType.OTHER :
+ CSAssignment.SkippedType.NONE;
CSAssignment assignment = new CSAssignment(skipped);
assignment.setApplication(application);
@@ -110,6 +112,11 @@ public abstract class AbstractContainerAllocator {
}
assignment.setContainersToKill(result.getToKillContainers());
+ } else {
+ if (result.getAllocationState() == AllocationState.QUEUE_SKIPPED) {
+ assignment.setSkippedType(
+ CSAssignment.SkippedType.QUEUE_LIMIT);
+ }
}
return assignment;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4d92aefd/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/IncreaseContainerAllocator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/IncreaseContainerAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/IncreaseContainerAllocator.java
index 4a2ae18..509dfba 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/IncreaseContainerAllocator.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/IncreaseContainerAllocator.java
@@ -71,7 +71,7 @@ public class IncreaseContainerAllocator extends AbstractContainerAllocator {
SchedContainerChangeRequest request) {
CSAssignment assignment =
new CSAssignment(request.getDeltaCapacity(), NodeType.NODE_LOCAL, null,
- application, false, false);
+ application, CSAssignment.SkippedType.NONE, false);
Resources.addTo(assignment.getAssignmentInformation().getReserved(),
request.getDeltaCapacity());
assignment.getAssignmentInformation().incrReservations();
@@ -88,7 +88,7 @@ public class IncreaseContainerAllocator extends AbstractContainerAllocator {
SchedContainerChangeRequest request, boolean fromReservation) {
CSAssignment assignment =
new CSAssignment(request.getDeltaCapacity(), NodeType.NODE_LOCAL, null,
- application, false, fromReservation);
+ application, CSAssignment.SkippedType.NONE, fromReservation);
Resources.addTo(assignment.getAssignmentInformation().getAllocated(),
request.getDeltaCapacity());
assignment.getAssignmentInformation().incrAllocations();
@@ -311,7 +311,8 @@ public class IncreaseContainerAllocator extends AbstractContainerAllocator {
// Try to allocate the increase request
assigned =
allocateIncreaseRequest(node, clusterResource, increaseRequest);
- if (!assigned.getSkipped()) {
+ if (assigned.getSkippedType()
+ == CSAssignment.SkippedType.NONE) {
// When we don't skip this request, which means we either allocated
// OR reserved this request. We will break
break;
@@ -328,7 +329,8 @@ public class IncreaseContainerAllocator extends AbstractContainerAllocator {
}
// We may have allocated something
- if (assigned != null && !assigned.getSkipped()) {
+ if (assigned != null && assigned.getSkippedType()
+ == CSAssignment.SkippedType.NONE) {
break;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4d92aefd/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
index d3567f5..2da7adb 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
@@ -113,6 +113,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeResourceUpdateEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.
+ ContainerExpiredSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplication;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
@@ -166,6 +168,12 @@ public class TestCapacityScheduler {
private static final String B3 = B + ".b3";
private static float A_CAPACITY = 10.5f;
private static float B_CAPACITY = 89.5f;
+ private static final String P1 = CapacitySchedulerConfiguration.ROOT + ".p1";
+ private static final String P2 = CapacitySchedulerConfiguration.ROOT + ".p2";
+ private static final String X1 = P1 + ".x1";
+ private static final String X2 = P1 + ".x2";
+ private static final String Y1 = P2 + ".y1";
+ private static final String Y2 = P2 + ".y2";
private static float A1_CAPACITY = 30;
private static float A2_CAPACITY = 70;
private static float B1_CAPACITY = 79.2f;
@@ -411,7 +419,52 @@ public class TestCapacityScheduler {
LOG.info("Setup top-level queues a and b");
return conf;
}
-
+
+ private CapacitySchedulerConfiguration setupBlockedQueueConfiguration(
+ CapacitySchedulerConfiguration conf) {
+
+ // Define top-level queues
+ conf.setQueues(CapacitySchedulerConfiguration.ROOT,
+ new String[]{"a", "b"});
+
+ conf.setCapacity(A, 80f);
+ conf.setCapacity(B, 20f);
+ conf.setUserLimitFactor(A, 100);
+ conf.setUserLimitFactor(B, 100);
+ conf.setMaximumCapacity(A, 100);
+ conf.setMaximumCapacity(B, 100);
+ LOG.info("Setup top-level queues a and b");
+ return conf;
+ }
+
+ private CapacitySchedulerConfiguration setupOtherBlockedQueueConfiguration(
+ CapacitySchedulerConfiguration conf) {
+
+ // Define top-level queues
+ conf.setQueues(CapacitySchedulerConfiguration.ROOT,
+ new String[]{"p1", "p2"});
+
+ conf.setCapacity(P1, 50f);
+ conf.setMaximumCapacity(P1, 50f);
+ conf.setCapacity(P2, 50f);
+ conf.setMaximumCapacity(P2, 100f);
+ // Define 2nd-level queues
+ conf.setQueues(P1, new String[] {"x1", "x2"});
+ conf.setCapacity(X1, 80f);
+ conf.setMaximumCapacity(X1, 100f);
+ conf.setUserLimitFactor(X1, 2f);
+ conf.setCapacity(X2, 20f);
+ conf.setMaximumCapacity(X2, 100f);
+ conf.setUserLimitFactor(X2, 2f);
+
+ conf.setQueues(P2, new String[]{"y1", "y2"});
+ conf.setCapacity(Y1, 80f);
+ conf.setUserLimitFactor(Y1, 2f);
+ conf.setCapacity(Y2, 20f);
+ conf.setUserLimitFactor(Y2, 2f);
+ return conf;
+ }
+
@Test
public void testMaximumCapacitySetup() {
float delta = 0.0000001f;
@@ -3415,4 +3468,237 @@ public class TestCapacityScheduler {
scheduler.handle(appRemovedEvent1);
rm.stop();
}
+
+ @Test
+ public void testCSReservationWithRootUnblocked() throws Exception {
+ CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
+ conf.setResourceComparator(DominantResourceCalculator.class);
+ setupOtherBlockedQueueConfiguration(conf);
+ conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
+ ResourceScheduler.class);
+ MockRM rm = new MockRM(conf);
+ rm.start();
+ CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
+ ParentQueue q = (ParentQueue) cs.getQueue("p1");
+
+ Assert.assertNotNull(q);
+ String host = "127.0.0.1";
+ String host1 = "test";
+ RMNode node =
+ MockNodes.newNodeInfo(0, Resource.newInstance(8 * GB, 8), 1, host);
+ RMNode node1 =
+ MockNodes.newNodeInfo(0, Resource.newInstance(8 * GB, 8), 2, host1);
+ cs.handle(new NodeAddedSchedulerEvent(node));
+ cs.handle(new NodeAddedSchedulerEvent(node1));
+ ApplicationAttemptId appAttemptId1 =
+ appHelper(rm, cs, 100, 1, "x1", "userX1");
+ ApplicationAttemptId appAttemptId2 =
+ appHelper(rm, cs, 100, 2, "x2", "userX2");
+ ApplicationAttemptId appAttemptId3 =
+ appHelper(rm, cs, 100, 3, "y1", "userY1");
+ RecordFactory recordFactory =
+ RecordFactoryProvider.getRecordFactory(null);
+
+ Priority priority = TestUtils.createMockPriority(1);
+ ResourceRequest y1Req = null;
+ ResourceRequest x1Req = null;
+ ResourceRequest x2Req = null;
+ for(int i=0; i < 4; i++) {
+ y1Req = TestUtils.createResourceRequest(
+ ResourceRequest.ANY, 1 * GB, 1, true, priority, recordFactory);
+ cs.allocate(appAttemptId3,
+ Collections.<ResourceRequest>singletonList(y1Req),
+ Collections.<ContainerId>emptyList(),
+ null, null, null, null);
+ CapacityScheduler.schedule(cs);
+ }
+ assertEquals("Y1 Used Resource should be 4 GB", 4 * GB,
+ cs.getQueue("y1").getUsedResources().getMemorySize());
+ assertEquals("P2 Used Resource should be 4 GB", 4 * GB,
+ cs.getQueue("p2").getUsedResources().getMemorySize());
+
+ for(int i=0; i < 7; i++) {
+ x1Req = TestUtils.createResourceRequest(
+ ResourceRequest.ANY, 1 * GB, 1, true, priority, recordFactory);
+ cs.allocate(appAttemptId1,
+ Collections.<ResourceRequest>singletonList(x1Req),
+ Collections.<ContainerId>emptyList(),
+ null, null, null, null);
+ CapacityScheduler.schedule(cs);
+ }
+ assertEquals("X1 Used Resource should be 7 GB", 7 * GB,
+ cs.getQueue("x1").getUsedResources().getMemorySize());
+ assertEquals("P1 Used Resource should be 7 GB", 7 * GB,
+ cs.getQueue("p1").getUsedResources().getMemorySize());
+
+ x2Req = TestUtils.createResourceRequest(
+ ResourceRequest.ANY, 2 * GB, 1, true, priority, recordFactory);
+ cs.allocate(appAttemptId2,
+ Collections.<ResourceRequest>singletonList(x2Req),
+ Collections.<ContainerId>emptyList(),
+ null, null, null, null);
+ CapacityScheduler.schedule(cs);
+ assertEquals("X2 Used Resource should be 0", 0,
+ cs.getQueue("x2").getUsedResources().getMemorySize());
+ assertEquals("P1 Used Resource should be 7 GB", 7 * GB,
+ cs.getQueue("p1").getUsedResources().getMemorySize());
+ //this assign should fail
+ x1Req = TestUtils.createResourceRequest(
+ ResourceRequest.ANY, 1 * GB, 1, true, priority, recordFactory);
+ cs.allocate(appAttemptId1,
+ Collections.<ResourceRequest>singletonList(x1Req),
+ Collections.<ContainerId>emptyList(),
+ null, null, null, null);
+ CapacityScheduler.schedule(cs);
+ assertEquals("X1 Used Resource should be 7 GB", 7 * GB,
+ cs.getQueue("x1").getUsedResources().getMemorySize());
+ assertEquals("P1 Used Resource should be 7 GB", 7 * GB,
+ cs.getQueue("p1").getUsedResources().getMemorySize());
+
+ //this should get thru
+ for (int i=0; i < 4; i++) {
+ y1Req = TestUtils.createResourceRequest(
+ ResourceRequest.ANY, 1 * GB, 1, true, priority, recordFactory);
+ cs.allocate(appAttemptId3,
+ Collections.<ResourceRequest>singletonList(y1Req),
+ Collections.<ContainerId>emptyList(),
+ null, null, null, null);
+ CapacityScheduler.schedule(cs);
+ }
+ assertEquals("P2 Used Resource should be 8 GB", 8 * GB,
+ cs.getQueue("p2").getUsedResources().getMemorySize());
+
+ //Free a container from X1
+ ContainerId containerId = ContainerId.newContainerId(appAttemptId1, 2);
+ cs.handle(new ContainerExpiredSchedulerEvent(containerId));
+
+ //Schedule pending request
+ CapacityScheduler.schedule(cs);
+ assertEquals("X2 Used Resource should be 2 GB", 2 * GB,
+ cs.getQueue("x2").getUsedResources().getMemorySize());
+ assertEquals("P1 Used Resource should be 8 GB", 8 * GB,
+ cs.getQueue("p1").getUsedResources().getMemorySize());
+ assertEquals("P2 Used Resource should be 8 GB", 8 * GB,
+ cs.getQueue("p2").getUsedResources().getMemorySize());
+ assertEquals("Root Used Resource should be 16 GB", 16 * GB,
+ cs.getRootQueue().getUsedResources().getMemorySize());
+ rm.stop();
+ }
+
+ @Test
+ public void testCSQueueBlocked() throws Exception {
+ CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
+ setupBlockedQueueConfiguration(conf);
+ conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
+ ResourceScheduler.class);
+ MockRM rm = new MockRM(conf);
+ rm.start();
+ CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
+ LeafQueue q = (LeafQueue) cs.getQueue("a");
+
+ Assert.assertNotNull(q);
+ String host = "127.0.0.1";
+ String host1 = "test";
+ RMNode node =
+ MockNodes.newNodeInfo(0, Resource.newInstance(8 * GB, 8), 1, host);
+ RMNode node1 =
+ MockNodes.newNodeInfo(0, Resource.newInstance(8 * GB, 8), 2, host1);
+ cs.handle(new NodeAddedSchedulerEvent(node));
+ cs.handle(new NodeAddedSchedulerEvent(node1));
+ //add app begin
+ ApplicationAttemptId appAttemptId1 =
+ appHelper(rm, cs, 100, 1, "a", "user1");
+ ApplicationAttemptId appAttemptId2 =
+ appHelper(rm, cs, 100, 2, "b", "user2");
+ //add app end
+
+ RecordFactory recordFactory =
+ RecordFactoryProvider.getRecordFactory(null);
+
+ Priority priority = TestUtils.createMockPriority(1);
+ ResourceRequest r1 = TestUtils.createResourceRequest(
+ ResourceRequest.ANY, 2 * GB, 1, true, priority, recordFactory);
+ //This will allocate for app1
+ cs.allocate(appAttemptId1, Collections.<ResourceRequest>singletonList(r1),
+ Collections.<ContainerId>emptyList(),
+ null, null, null, null).getContainers().size();
+ CapacityScheduler.schedule(cs);
+ ResourceRequest r2 = null;
+ for (int i =0; i < 13; i++) {
+ r2 = TestUtils.createResourceRequest(
+ ResourceRequest.ANY, 1 * GB, 1, true, priority, recordFactory);
+ cs.allocate(appAttemptId2,
+ Collections.<ResourceRequest>singletonList(r2),
+ Collections.<ContainerId>emptyList(),
+ null, null, null, null);
+ CapacityScheduler.schedule(cs);
+ }
+ assertEquals("A Used Resource should be 2 GB", 2 * GB,
+ cs.getQueue("a").getUsedResources().getMemorySize());
+ assertEquals("B Used Resource should be 2 GB", 13 * GB,
+ cs.getQueue("b").getUsedResources().getMemorySize());
+ r1 = TestUtils.createResourceRequest(
+ ResourceRequest.ANY, 2 * GB, 1, true, priority, recordFactory);
+ r2 = TestUtils.createResourceRequest(
+ ResourceRequest.ANY, 1 * GB, 1, true, priority, recordFactory);
+ cs.allocate(appAttemptId1, Collections.<ResourceRequest>singletonList(r1),
+ Collections.<ContainerId>emptyList(),
+ null, null, null, null).getContainers().size();
+ CapacityScheduler.schedule(cs);
+
+ cs.allocate(appAttemptId2, Collections.<ResourceRequest>singletonList(r2),
+ Collections.<ContainerId>emptyList(), null, null, null, null);
+ CapacityScheduler.schedule(cs);
+ //Check blocked Resource
+ assertEquals("A Used Resource should be 2 GB", 2 * GB,
+ cs.getQueue("a").getUsedResources().getMemorySize());
+ assertEquals("B Used Resource should be 13 GB", 13 * GB,
+ cs.getQueue("b").getUsedResources().getMemorySize());
+
+ ContainerId containerId1 = ContainerId.newContainerId(appAttemptId2, 10);
+ ContainerId containerId2 =ContainerId.newContainerId(appAttemptId2, 11);
+
+ cs.handle(new ContainerExpiredSchedulerEvent(containerId1));
+ cs.handle(new ContainerExpiredSchedulerEvent(containerId2));
+ CapacityScheduler.schedule(cs);
+ rm.drainEvents();
+ assertEquals("A Used Resource should be 2 GB", 4 * GB,
+ cs.getQueue("a").getUsedResources().getMemorySize());
+ assertEquals("B Used Resource should be 12 GB", 12 * GB,
+ cs.getQueue("b").getUsedResources().getMemorySize());
+ assertEquals("Used Resource on Root should be 16 GB", 16 * GB,
+ cs.getRootQueue().getUsedResources().getMemorySize());
+ rm.stop();
+ }
+
+ private ApplicationAttemptId appHelper(MockRM rm, CapacityScheduler cs,
+ int clusterTs, int appId, String queue,
+ String user) {
+ ApplicationId appId1 = BuilderUtils.newApplicationId(clusterTs, appId);
+ ApplicationAttemptId appAttemptId1 = BuilderUtils.newApplicationAttemptId(
+ appId1, appId);
+
+ RMAppAttemptMetrics attemptMetric1 =
+ new RMAppAttemptMetrics(appAttemptId1, rm.getRMContext());
+ RMAppImpl app1 = mock(RMAppImpl.class);
+ when(app1.getApplicationId()).thenReturn(appId1);
+ RMAppAttemptImpl attempt1 = mock(RMAppAttemptImpl.class);
+ Container container = mock(Container.class);
+ when(attempt1.getMasterContainer()).thenReturn(container);
+ ApplicationSubmissionContext submissionContext = mock(
+ ApplicationSubmissionContext.class);
+ when(attempt1.getSubmissionContext()).thenReturn(submissionContext);
+ when(attempt1.getAppAttemptId()).thenReturn(appAttemptId1);
+ when(attempt1.getRMAppAttemptMetrics()).thenReturn(attemptMetric1);
+ when(app1.getCurrentAppAttempt()).thenReturn(attempt1);
+ rm.getRMContext().getRMApps().put(appId1, app1);
+
+ SchedulerEvent addAppEvent1 =
+ new AppAddedSchedulerEvent(appId1, queue, user);
+ cs.handle(addAppEvent1);
+ SchedulerEvent addAttemptEvent1 =
+ new AppAttemptAddedSchedulerEvent(appAttemptId1, false);
+ cs.handle(addAttemptEvent1);
+ return appAttemptId1;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[19/30] hadoop git commit: HADOOP-13483. Optimize IPC server protobuf
decoding. Contributed by Daryn Sharp.
Posted by dr...@apache.org.
HADOOP-13483. Optimize IPC server protobuf decoding. Contributed by Daryn Sharp.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/580a8334
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/580a8334
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/580a8334
Branch: refs/heads/HADOOP-12756
Commit: 580a8334963709e728ed677c815fb7fef9bca70e
Parents: 22ef528
Author: Kihwal Lee <ki...@apache.org>
Authored: Wed Aug 3 13:22:22 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Wed Aug 3 13:22:22 2016 -0500
----------------------------------------------------------------------
.../apache/hadoop/ipc/ProtobufRpcEngine.java | 18 +-
.../java/org/apache/hadoop/ipc/RpcWritable.java | 184 +++++++++++++++++++
.../main/java/org/apache/hadoop/ipc/Server.java | 94 ++++------
.../org/apache/hadoop/ipc/TestRpcWritable.java | 129 +++++++++++++
4 files changed, 362 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/580a8334/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
index 315ec67..cce5166 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
@@ -32,6 +32,7 @@ import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.ipc.Client.ConnectionId;
import org.apache.hadoop.ipc.RPC.RpcInvoker;
+import org.apache.hadoop.ipc.RpcWritable;
import org.apache.hadoop.ipc.protobuf.ProtobufRpcEngineProtos.RequestHeaderProto;
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto;
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
@@ -68,7 +69,7 @@ public class ProtobufRpcEngine implements RpcEngine {
static { // Register the rpcRequest deserializer for WritableRpcEngine
org.apache.hadoop.ipc.Server.registerProtocolEngine(
- RPC.RpcKind.RPC_PROTOCOL_BUFFER, RpcRequestWrapper.class,
+ RPC.RpcKind.RPC_PROTOCOL_BUFFER, RpcWritable.Buffer.class,
new Server.ProtoBufRpcInvoker());
}
@@ -612,11 +613,11 @@ public class ProtobufRpcEngine implements RpcEngine {
*/
public Writable call(RPC.Server server, String connectionProtocolName,
Writable writableRequest, long receiveTime) throws Exception {
- RpcRequestWrapper request = (RpcRequestWrapper) writableRequest;
- RequestHeaderProto rpcRequest = request.requestHeader;
+ RpcWritable.Buffer request = (RpcWritable.Buffer) writableRequest;
+ RequestHeaderProto rpcRequest =
+ request.getValue(RequestHeaderProto.getDefaultInstance());
String methodName = rpcRequest.getMethodName();
-
-
+
/**
* RPCs for a particular interface (ie protocol) are done using a
* IPC connection that is setup using rpcProxy.
@@ -652,9 +653,8 @@ public class ProtobufRpcEngine implements RpcEngine {
throw new RpcNoSuchMethodException(msg);
}
Message prototype = service.getRequestPrototype(methodDescriptor);
- Message param = prototype.newBuilderForType()
- .mergeFrom(request.theRequestRead).build();
-
+ Message param = request.getValue(prototype);
+
Message result;
long startTime = Time.now();
int qTime = (int) (startTime - receiveTime);
@@ -683,7 +683,7 @@ public class ProtobufRpcEngine implements RpcEngine {
exception.getClass().getSimpleName();
server.updateMetrics(detailedMetricsName, qTime, processingTime);
}
- return new RpcResponseWrapper(result);
+ return RpcWritable.wrap(result);
}
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/580a8334/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcWritable.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcWritable.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcWritable.java
new file mode 100644
index 0000000..5125939
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcWritable.java
@@ -0,0 +1,184 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ipc;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Writable;
+
+import com.google.protobuf.CodedInputStream;
+import com.google.protobuf.CodedOutputStream;
+import com.google.protobuf.Message;
+
+@InterfaceAudience.Private
+public abstract class RpcWritable implements Writable {
+
+ static RpcWritable wrap(Object o) {
+ if (o instanceof RpcWritable) {
+ return (RpcWritable)o;
+ } else if (o instanceof Message) {
+ return new ProtobufWrapper((Message)o);
+ } else if (o instanceof Writable) {
+ return new WritableWrapper((Writable)o);
+ }
+ throw new IllegalArgumentException("Cannot wrap " + o.getClass());
+ }
+
+ // don't support old inefficient Writable methods.
+ @Override
+ public final void readFields(DataInput in) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+ @Override
+ public final void write(DataOutput out) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ // methods optimized for reduced intermediate byte[] allocations.
+ abstract void writeTo(ResponseBuffer out) throws IOException;
+ abstract <T> T readFrom(ByteBuffer bb) throws IOException;
+
+ // adapter for Writables.
+ static class WritableWrapper extends RpcWritable {
+ private final Writable writable;
+
+ WritableWrapper(Writable writable) {
+ this.writable = writable;
+ }
+
+ @Override
+ public void writeTo(ResponseBuffer out) throws IOException {
+ writable.write(out);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ <T> T readFrom(ByteBuffer bb) throws IOException {
+ // create a stream that may consume up to the entire ByteBuffer.
+ DataInputStream in = new DataInputStream(new ByteArrayInputStream(
+ bb.array(), bb.position() + bb.arrayOffset(), bb.remaining()));
+ try {
+ writable.readFields(in);
+ } finally {
+ // advance over the bytes read.
+ bb.position(bb.limit() - in.available());
+ }
+ return (T)writable;
+ }
+ }
+
+ // adapter for Protobufs.
+ static class ProtobufWrapper extends RpcWritable {
+ private Message message;
+
+ ProtobufWrapper(Message message) {
+ this.message = message;
+ }
+
+ @Override
+ void writeTo(ResponseBuffer out) throws IOException {
+ int length = message.getSerializedSize();
+ length += CodedOutputStream.computeRawVarint32Size(length);
+ out.ensureCapacity(length);
+ message.writeDelimitedTo(out);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ <T> T readFrom(ByteBuffer bb) throws IOException {
+ // using the parser with a byte[]-backed coded input stream is the
+ // most efficient way to deserialize a protobuf. it has a direct
+ // path to the PB ctor that doesn't create multi-layered streams
+ // that internally buffer.
+ CodedInputStream cis = CodedInputStream.newInstance(
+ bb.array(), bb.position() + bb.arrayOffset(), bb.remaining());
+ try {
+ cis.pushLimit(cis.readRawVarint32());
+ message = message.getParserForType().parseFrom(cis);
+ cis.checkLastTagWas(0);
+ } finally {
+ // advance over the bytes read.
+ bb.position(bb.position() + cis.getTotalBytesRead());
+ }
+ return (T)message;
+ }
+ }
+
+ // adapter to allow decoding of writables and protobufs from a byte buffer.
+ static class Buffer extends RpcWritable {
+ private ByteBuffer bb;
+
+ static Buffer wrap(ByteBuffer bb) {
+ return new Buffer(bb);
+ }
+
+ Buffer() {}
+
+ Buffer(ByteBuffer bb) {
+ this.bb = bb;
+ }
+
+ @Override
+ void writeTo(ResponseBuffer out) throws IOException {
+ out.ensureCapacity(bb.remaining());
+ out.write(bb.array(), bb.position() + bb.arrayOffset(), bb.remaining());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ <T> T readFrom(ByteBuffer bb) throws IOException {
+ // effectively consume the rest of the buffer from the callers
+ // perspective.
+ this.bb = bb.slice();
+ bb.limit(bb.position());
+ return (T)this;
+ }
+
+ public <T> T newInstance(Class<T> valueClass,
+ Configuration conf) throws IOException {
+ T instance;
+ try {
+ // this is much faster than ReflectionUtils!
+ instance = valueClass.newInstance();
+ if (instance instanceof Configurable) {
+ ((Configurable)instance).setConf(conf);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return getValue(instance);
+ }
+
+ public <T> T getValue(T value) throws IOException {
+ return RpcWritable.wrap(value).readFrom(bb);
+ }
+
+ int remaining() {
+ return bb.remaining();
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/580a8334/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
index 4004fc0..4c73f6a 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
@@ -26,7 +26,6 @@ import static org.apache.hadoop.ipc.RpcConstants.PING_CALL_ID;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.lang.reflect.UndeclaredThrowableException;
@@ -83,8 +82,6 @@ import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
-import org.apache.hadoop.ipc.ProtobufRpcEngine.RpcResponseWrapper;
-import org.apache.hadoop.ipc.ProtobufRpcEngine.RpcWrapper;
import org.apache.hadoop.ipc.RPC.RpcInvoker;
import org.apache.hadoop.ipc.RPC.VersionMismatch;
import org.apache.hadoop.ipc.metrics.RpcDetailedMetrics;
@@ -114,7 +111,6 @@ import org.apache.hadoop.security.token.SecretManager;
import org.apache.hadoop.security.token.SecretManager.InvalidToken;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.hadoop.util.ProtoUtil;
-import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Time;
import org.apache.htrace.core.SpanId;
@@ -123,9 +119,7 @@ import org.apache.htrace.core.Tracer;
import com.google.common.annotations.VisibleForTesting;
import com.google.protobuf.ByteString;
-import com.google.protobuf.CodedOutputStream;
import com.google.protobuf.Message;
-import com.google.protobuf.Message.Builder;
/** An abstract IPC service. IPC calls take a single {@link Writable} as a
* parameter, and return a {@link Writable} as their value. A service runs on
@@ -1346,6 +1340,7 @@ public abstract class Server {
* A WrappedRpcServerException that is suppressed altogether
* for the purposes of logging.
*/
+ @SuppressWarnings("serial")
private static class WrappedRpcServerExceptionSuppressed
extends WrappedRpcServerException {
public WrappedRpcServerExceptionSuppressed(
@@ -1478,10 +1473,10 @@ public abstract class Server {
}
}
- private void saslReadAndProcess(DataInputStream dis) throws
+ private void saslReadAndProcess(RpcWritable.Buffer buffer) throws
WrappedRpcServerException, IOException, InterruptedException {
final RpcSaslProto saslMessage =
- decodeProtobufFromStream(RpcSaslProto.newBuilder(), dis);
+ getMessage(RpcSaslProto.getDefaultInstance(), buffer);
switch (saslMessage.getState()) {
case WRAP: {
if (!saslContextEstablished || !useWrap) {
@@ -1713,7 +1708,7 @@ public abstract class Server {
RpcConstants.INVALID_RETRY_COUNT, null, this);
setupResponse(saslCall,
RpcStatusProto.SUCCESS, null,
- new RpcResponseWrapper(message), null, null);
+ RpcWritable.wrap(message), null, null);
saslCall.sendResponse();
}
@@ -1839,7 +1834,7 @@ public abstract class Server {
dataLengthBuffer.clear(); // to read length of future rpc packets
data.flip();
boolean isHeaderRead = connectionContextRead;
- processOneRpc(data.array());
+ processOneRpc(data);
data = null;
// the last rpc-request we processed could have simply been the
// connectionContext; if so continue to read the first RPC.
@@ -1966,7 +1961,7 @@ public abstract class Server {
* @throws WrappedRpcServerException - if the header cannot be
* deserialized, or the user is not authorized
*/
- private void processConnectionContext(DataInputStream dis)
+ private void processConnectionContext(RpcWritable.Buffer buffer)
throws WrappedRpcServerException {
// allow only one connection context during a session
if (connectionContextRead) {
@@ -1974,8 +1969,7 @@ public abstract class Server {
RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
"Connection context already processed");
}
- connectionContext = decodeProtobufFromStream(
- IpcConnectionContextProto.newBuilder(), dis);
+ connectionContext = getMessage(IpcConnectionContextProto.getDefaultInstance(), buffer);
protocolName = connectionContext.hasProtocol() ? connectionContext
.getProtocol() : null;
@@ -2053,7 +2047,7 @@ public abstract class Server {
if (unwrappedData.remaining() == 0) {
unwrappedDataLengthBuffer.clear();
unwrappedData.flip();
- processOneRpc(unwrappedData.array());
+ processOneRpc(unwrappedData);
unwrappedData = null;
}
}
@@ -2077,31 +2071,30 @@ public abstract class Server {
* client in this method and does not require verbose logging by the
* Listener thread
* @throws InterruptedException
- */
- private void processOneRpc(byte[] buf)
+ */
+ private void processOneRpc(ByteBuffer bb)
throws IOException, WrappedRpcServerException, InterruptedException {
int callId = -1;
int retry = RpcConstants.INVALID_RETRY_COUNT;
try {
- final DataInputStream dis =
- new DataInputStream(new ByteArrayInputStream(buf));
+ final RpcWritable.Buffer buffer = RpcWritable.Buffer.wrap(bb);
final RpcRequestHeaderProto header =
- decodeProtobufFromStream(RpcRequestHeaderProto.newBuilder(), dis);
+ getMessage(RpcRequestHeaderProto.getDefaultInstance(), buffer);
callId = header.getCallId();
retry = header.getRetryCount();
if (LOG.isDebugEnabled()) {
LOG.debug(" got #" + callId);
}
checkRpcHeaders(header);
-
+
if (callId < 0) { // callIds typically used during connection setup
- processRpcOutOfBandRequest(header, dis);
+ processRpcOutOfBandRequest(header, buffer);
} else if (!connectionContextRead) {
throw new WrappedRpcServerException(
RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
"Connection context not established");
} else {
- processRpcRequest(header, dis);
+ processRpcRequest(header, buffer);
}
} catch (WrappedRpcServerException wrse) { // inform client of error
Throwable ioe = wrse.getCause();
@@ -2157,7 +2150,7 @@ public abstract class Server {
* @throws InterruptedException
*/
private void processRpcRequest(RpcRequestHeaderProto header,
- DataInputStream dis) throws WrappedRpcServerException,
+ RpcWritable.Buffer buffer) throws WrappedRpcServerException,
InterruptedException {
Class<? extends Writable> rpcRequestClass =
getRpcRequestWrapper(header.getRpcKind());
@@ -2171,8 +2164,7 @@ public abstract class Server {
}
Writable rpcRequest;
try { //Read the rpc request
- rpcRequest = ReflectionUtils.newInstance(rpcRequestClass, conf);
- rpcRequest.readFields(dis);
+ rpcRequest = buffer.newInstance(rpcRequestClass, conf);
} catch (Throwable t) { // includes runtime exception from newInstance
LOG.warn("Unable to read call parameters for client " +
getHostAddress() + "on connection protocol " +
@@ -2253,8 +2245,8 @@ public abstract class Server {
* @throws InterruptedException
*/
private void processRpcOutOfBandRequest(RpcRequestHeaderProto header,
- DataInputStream dis) throws WrappedRpcServerException, IOException,
- InterruptedException {
+ RpcWritable.Buffer buffer) throws WrappedRpcServerException,
+ IOException, InterruptedException {
final int callId = header.getCallId();
if (callId == CONNECTION_CONTEXT_CALL_ID) {
// SASL must be established prior to connection context
@@ -2264,7 +2256,7 @@ public abstract class Server {
"Connection header sent during SASL negotiation");
}
// read and authorize the user
- processConnectionContext(dis);
+ processConnectionContext(buffer);
} else if (callId == AuthProtocol.SASL.callId) {
// if client was switched to simple, ignore first SASL message
if (authProtocol != AuthProtocol.SASL) {
@@ -2272,7 +2264,7 @@ public abstract class Server {
RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
"SASL protocol not requested by client");
}
- saslReadAndProcess(dis);
+ saslReadAndProcess(buffer);
} else if (callId == PING_CALL_ID) {
LOG.debug("Received ping message");
} else {
@@ -2319,13 +2311,12 @@ public abstract class Server {
* @throws WrappedRpcServerException - deserialization failed
*/
@SuppressWarnings("unchecked")
- private <T extends Message> T decodeProtobufFromStream(Builder builder,
- DataInputStream dis) throws WrappedRpcServerException {
+ <T extends Message> T getMessage(Message message,
+ RpcWritable.Buffer buffer) throws WrappedRpcServerException {
try {
- builder.mergeDelimitedFrom(dis);
- return (T)builder.build();
+ return (T)buffer.getValue(message);
} catch (Exception ioe) {
- Class<?> protoClass = builder.getDefaultInstanceForType().getClass();
+ Class<?> protoClass = message.getClass();
throw new WrappedRpcServerException(
RpcErrorCodeProto.FATAL_DESERIALIZING_REQUEST,
"Error decoding " + protoClass.getSimpleName() + ": "+ ioe);
@@ -2716,25 +2707,20 @@ public abstract class Server {
private void setupResponse(Call call,
RpcResponseHeaderProto header, Writable rv) throws IOException {
ResponseBuffer buf = responseBuffer.get().reset();
- // adjust capacity on estimated length to reduce resizing copies
- int estimatedLen = header.getSerializedSize();
- estimatedLen += CodedOutputStream.computeRawVarint32Size(estimatedLen);
- // if it's not a wrapped protobuf, just let it grow on its own
- if (rv instanceof RpcWrapper) {
- estimatedLen += ((RpcWrapper)rv).getLength();
- }
- buf.ensureCapacity(estimatedLen);
- header.writeDelimitedTo(buf);
- if (rv != null) { // null for exceptions
- rv.write(buf);
- }
- call.setResponse(ByteBuffer.wrap(buf.toByteArray()));
- // Discard a large buf and reset it back to smaller size
- // to free up heap.
- if (buf.capacity() > maxRespSize) {
- LOG.warn("Large response size " + buf.size() + " for call "
- + call.toString());
- buf.setCapacity(INITIAL_RESP_BUF_SIZE);
+ try {
+ RpcWritable.wrap(header).writeTo(buf);
+ if (rv != null) {
+ RpcWritable.wrap(rv).writeTo(buf);
+ }
+ call.setResponse(ByteBuffer.wrap(buf.toByteArray()));
+ } finally {
+ // Discard a large buf and reset it back to smaller size
+ // to free up heap.
+ if (buf.capacity() > maxRespSize) {
+ LOG.warn("Large response size " + buf.size() + " for call "
+ + call.toString());
+ buf.setCapacity(INITIAL_RESP_BUF_SIZE);
+ }
}
}
@@ -2785,7 +2771,7 @@ public abstract class Server {
.setState(SaslState.WRAP)
.setToken(ByteString.copyFrom(token))
.build();
- setupResponse(call, saslHeader, new RpcResponseWrapper(saslMessage));
+ setupResponse(call, saslHeader, RpcWritable.wrap(saslMessage));
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/580a8334/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRpcWritable.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRpcWritable.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRpcWritable.java
new file mode 100644
index 0000000..837f579
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRpcWritable.java
@@ -0,0 +1,129 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ipc;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto;
+import org.apache.hadoop.util.Time;
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.google.protobuf.Message;
+
+public class TestRpcWritable {//extends TestRpcBase {
+
+ static Writable writable = new LongWritable(Time.now());
+ static Message message1 =
+ EchoRequestProto.newBuilder().setMessage("testing1").build();
+ static Message message2 =
+ EchoRequestProto.newBuilder().setMessage("testing2").build();
+
+ @Test
+ public void testWritableWrapper() throws IOException {
+ // serial writable in byte buffer
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ writable.write(new DataOutputStream(baos));
+ ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+
+ // deserial
+ LongWritable actual = RpcWritable.wrap(new LongWritable())
+ .readFrom(bb);
+ Assert.assertEquals(writable, actual);
+ Assert.assertEquals(0, bb.remaining());
+ }
+
+ @Test
+ public void testProtobufWrapper() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ message1.writeDelimitedTo(baos);
+ ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+
+ Message actual = RpcWritable.wrap(EchoRequestProto.getDefaultInstance())
+ .readFrom(bb);
+ Assert.assertEquals(message1, actual);
+ Assert.assertEquals(0, bb.remaining());
+ }
+
+ @Test
+ public void testBufferWrapper() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(baos);
+ message1.writeDelimitedTo(dos);
+ message2.writeDelimitedTo(dos);
+ writable.write(dos);
+
+ ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+ RpcWritable.Buffer buf = RpcWritable.Buffer.wrap(bb);
+ Assert.assertEquals(baos.size(), bb.remaining());
+ Assert.assertEquals(baos.size(), buf.remaining());
+
+ Object actual = buf.getValue(EchoRequestProto.getDefaultInstance());
+ Assert.assertEquals(message1, actual);
+ Assert.assertTrue(bb.remaining() > 0);
+ Assert.assertEquals(bb.remaining(), buf.remaining());
+
+ actual = buf.getValue(EchoRequestProto.getDefaultInstance());
+ Assert.assertEquals(message2, actual);
+ Assert.assertTrue(bb.remaining() > 0);
+ Assert.assertEquals(bb.remaining(), buf.remaining());
+
+ actual = buf.newInstance(LongWritable.class, null);
+ Assert.assertEquals(writable, actual);
+ Assert.assertEquals(0, bb.remaining());
+ Assert.assertEquals(0, buf.remaining());
+ }
+
+ @Test
+ public void testBufferWrapperNested() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(baos);
+ writable.write(dos);
+ message1.writeDelimitedTo(dos);
+ message2.writeDelimitedTo(dos);
+ ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+ RpcWritable.Buffer buf1 = RpcWritable.Buffer.wrap(bb);
+ Assert.assertEquals(baos.size(), bb.remaining());
+ Assert.assertEquals(baos.size(), buf1.remaining());
+
+ Object actual = buf1.newInstance(LongWritable.class, null);
+ Assert.assertEquals(writable, actual);
+ int left = bb.remaining();
+ Assert.assertTrue(left > 0);
+ Assert.assertEquals(left, buf1.remaining());
+
+ // original bb now appears empty, but rpc writable has a slice of the bb.
+ RpcWritable.Buffer buf2 = buf1.newInstance(RpcWritable.Buffer.class, null);
+ Assert.assertEquals(0, bb.remaining());
+ Assert.assertEquals(0, buf1.remaining());
+ Assert.assertEquals(left, buf2.remaining());
+
+ actual = buf2.getValue(EchoRequestProto.getDefaultInstance());
+ Assert.assertEquals(message1, actual);
+ Assert.assertTrue(buf2.remaining() > 0);
+
+ actual = buf2.getValue(EchoRequestProto.getDefaultInstance());
+ Assert.assertEquals(message2, actual);
+ Assert.assertEquals(0, buf2.remaining());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[17/30] hadoop git commit: HDFS-742. A down DataNode makes Balancer
to hang on repeatingly asking NameNode its partial block list. Contributed by
Mit Desai.
Posted by dr...@apache.org.
HDFS-742. A down DataNode makes Balancer to hang on repeatingly asking NameNode its partial block list. Contributed by Mit Desai.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/58db263e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/58db263e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/58db263e
Branch: refs/heads/HADOOP-12756
Commit: 58db263e93daf08280e6a586a10cebd6122cf72a
Parents: bebf10d
Author: Kihwal Lee <ki...@apache.org>
Authored: Wed Aug 3 12:14:59 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Wed Aug 3 12:14:59 2016 -0500
----------------------------------------------------------------------
.../hadoop/hdfs/server/balancer/Dispatcher.java | 15 +++++++--------
1 file changed, 7 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/58db263e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
index 9f9cdc0..16e42e0 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
@@ -849,6 +849,13 @@ public class Dispatcher {
+ ", scheduledSize=" + getScheduledSize()
+ ", srcBlocks#=" + srcBlocks.size());
}
+ // check if time is up or not
+ if (Time.monotonicNow() - startTime > MAX_ITERATION_TIME) {
+ LOG.info("Time up (max time=" + MAX_ITERATION_TIME/1000
+ + " seconds). Skipping " + this);
+ isTimeUp = true;
+ continue;
+ }
final PendingMove p = chooseNextMove();
if (p != null) {
// Reset no pending move counter
@@ -886,14 +893,6 @@ public class Dispatcher {
}
}
- // check if time is up or not
- if (Time.monotonicNow() - startTime > MAX_ITERATION_TIME) {
- LOG.info("Time up (max time=" + MAX_ITERATION_TIME/1000
- + " seconds). Skipping " + this);
- isTimeUp = true;
- continue;
- }
-
// Now we can not schedule any block to move and there are
// no new blocks added to the source block list, so we wait.
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[27/30] hadoop git commit: MAPREDUCE-6682.
TestMRCJCFileOutputCommitter fails intermittently Contributed by Akira
Ajisaka.
Posted by dr...@apache.org.
MAPREDUCE-6682. TestMRCJCFileOutputCommitter fails intermittently Contributed by Akira Ajisaka.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/8f1c374b
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/8f1c374b
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/8f1c374b
Branch: refs/heads/HADOOP-12756
Commit: 8f1c374bec2451568f463ab68f7eb6db734ab14e
Parents: a1f6564
Author: Brahma Reddy Battula <br...@apache.org>
Authored: Thu Aug 4 11:21:47 2016 +0530
Committer: Brahma Reddy Battula <br...@apache.org>
Committed: Thu Aug 4 11:21:47 2016 +0530
----------------------------------------------------------------------
.../mapred/TestMRCJCFileOutputCommitter.java | 14 +++++---
.../mapred/TestMRTimelineEventHandling.java | 34 +++++++++++++-------
2 files changed, 32 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f1c374b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRCJCFileOutputCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRCJCFileOutputCommitter.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRCJCFileOutputCommitter.java
index 74b6d77..83796e8 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRCJCFileOutputCommitter.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRCJCFileOutputCommitter.java
@@ -25,6 +25,8 @@ import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.JobStatus;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.After;
import org.junit.Test;
import java.io.File;
@@ -37,8 +39,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
public class TestMRCJCFileOutputCommitter {
- private static Path outDir = new Path(
- System.getProperty("test.build.data", "/tmp"), "output");
+ private static Path outDir = new Path(GenericTestUtils.getTempPath("output"));
// A random task attempt id for testing.
private static String attempt = "attempt_200707121733_0001_m_000000_0";
@@ -112,12 +113,11 @@ public class TestMRCJCFileOutputCommitter {
expectedOutput.append(key2).append('\t').append(val2).append("\n");
String output = UtilsForTests.slurp(expectedFile);
assertEquals(output, expectedOutput.toString());
-
- FileUtil.fullyDelete(new File(outDir.toString()));
}
@Test
public void testAbort() throws IOException {
+ FileUtil.fullyDelete(new File(outDir.toString()));
JobConf job = new JobConf();
setConfForFileOutputCommitter(job);
JobContext jContext = new JobContextImpl(job, taskID.getJobID());
@@ -152,7 +152,6 @@ public class TestMRCJCFileOutputCommitter {
assertFalse("job temp dir "+expectedFile+" still exists", expectedFile.exists());
assertEquals("Output directory not empty", 0, new File(outDir.toString())
.listFiles().length);
- FileUtil.fullyDelete(new File(outDir.toString()));
}
public static class FakeFileSystem extends RawLocalFileSystem {
@@ -223,4 +222,9 @@ public class TestMRCJCFileOutputCommitter {
assertTrue(th.getMessage().contains("fake delete failed"));
assertTrue("job temp dir does not exists", jobTmpDir.exists());
}
+
+ @After
+ public void teardown() {
+ FileUtil.fullyDelete(new File(outDir.toString()));
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f1c374b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRTimelineEventHandling.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRTimelineEventHandling.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRTimelineEventHandling.java
index 53ad6bc..cbca3c8 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRTimelineEventHandling.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRTimelineEventHandling.java
@@ -34,12 +34,14 @@ import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.jobhistory.EventType;
import org.apache.hadoop.mapreduce.jobhistory.TestJobHistoryEventHandler;
import org.apache.hadoop.mapreduce.v2.MiniMRYarnCluster;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
@@ -460,21 +462,21 @@ public class TestMRTimelineEventHandling {
conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
conf.setBoolean(MRJobConfig.MAPREDUCE_JOB_EMIT_TIMELINE_DATA, false);
MiniMRYarnCluster cluster = null;
+ FileSystem fs = null;
+ Path inDir = new Path(GenericTestUtils.getTempPath("input"));
+ Path outDir = new Path(GenericTestUtils.getTempPath("output"));
try {
+ fs = FileSystem.get(conf);
cluster = new MiniMRYarnCluster(
TestMRTimelineEventHandling.class.getSimpleName(), 1);
cluster.init(conf);
cluster.start();
conf.set(YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS,
MiniYARNCluster.getHostname() + ":"
- + cluster.getApplicationHistoryServer().getPort());
+ + cluster.getApplicationHistoryServer().getPort());
TimelineStore ts = cluster.getApplicationHistoryServer()
.getTimelineStore();
- String localPathRoot = System.getProperty("test.build.data",
- "build/test/data");
- Path inDir = new Path(localPathRoot, "input");
- Path outDir = new Path(localPathRoot, "output");
RunningJob job =
UtilsForTests.runJobSucceed(new JobConf(conf), inDir, outDir);
Assert.assertEquals(JobStatus.SUCCEEDED,
@@ -496,6 +498,7 @@ public class TestMRTimelineEventHandling {
if (cluster != null) {
cluster.stop();
}
+ deletePaths(fs, inDir, outDir);
}
conf = new YarnConfiguration();
@@ -509,15 +512,10 @@ public class TestMRTimelineEventHandling {
cluster.start();
conf.set(YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS,
MiniYARNCluster.getHostname() + ":"
- + cluster.getApplicationHistoryServer().getPort());
+ + cluster.getApplicationHistoryServer().getPort());
TimelineStore ts = cluster.getApplicationHistoryServer()
.getTimelineStore();
- String localPathRoot = System.getProperty("test.build.data",
- "build/test/data");
- Path inDir = new Path(localPathRoot, "input");
- Path outDir = new Path(localPathRoot, "output");
-
conf.setBoolean(MRJobConfig.MAPREDUCE_JOB_EMIT_TIMELINE_DATA, false);
RunningJob job =
UtilsForTests.runJobSucceed(new JobConf(conf), inDir, outDir);
@@ -540,6 +538,20 @@ public class TestMRTimelineEventHandling {
if (cluster != null) {
cluster.stop();
}
+ deletePaths(fs, inDir, outDir);
+ }
+ }
+
+ /** Delete input paths recursively. Paths should not be null. */
+ private void deletePaths(FileSystem fs, Path... paths) {
+ if (fs == null) {
+ return;
+ }
+ for (Path path : paths) {
+ try {
+ fs.delete(path, true);
+ } catch (Exception ignored) {
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[12/30] hadoop git commit: YARN-5456. container-executor support for
FreeBSD, NetBSD,
and others if conf path is absolute. Contributed by Allen Wittenauer.
Posted by dr...@apache.org.
YARN-5456. container-executor support for FreeBSD, NetBSD, and others if conf path is absolute. Contributed by Allen Wittenauer.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/b9136773
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/b9136773
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/b9136773
Branch: refs/heads/HADOOP-12756
Commit: b913677365ad77ca7daa5741c04c14df1a0313cd
Parents: d28c2d9
Author: Chris Nauroth <cn...@apache.org>
Authored: Tue Aug 2 22:24:34 2016 -0700
Committer: Chris Nauroth <cn...@apache.org>
Committed: Tue Aug 2 22:24:34 2016 -0700
----------------------------------------------------------------------
.../src/CMakeLists.txt | 13 ++
.../src/config.h.cmake | 12 ++
.../container-executor/impl/configuration.h | 4 +
.../impl/container-executor.c | 26 +++-
.../impl/container-executor.h | 8 +-
.../container-executor/impl/get_executable.c | 118 ++++++++++++++++---
.../main/native/container-executor/impl/main.c | 6 +-
.../test/test-container-executor.c | 4 +
8 files changed, 167 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9136773/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/CMakeLists.txt b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/CMakeLists.txt
index 204b3ca..fbc794c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/CMakeLists.txt
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/CMakeLists.txt
@@ -19,10 +19,20 @@ cmake_minimum_required(VERSION 2.6 FATAL_ERROR)
list(APPEND CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/../../../../../hadoop-common-project/hadoop-common)
include(HadoopCommon)
+# determine if container-executor.conf.dir is an absolute
+# path in case the OS we're compiling on doesn't have
+# a hook in get_executable. We'll use this define
+# later in the code to potentially throw a compile error
+string(REGEX MATCH . HCD_ONE "${HADOOP_CONF_DIR}")
+string(COMPARE EQUAL ${HCD_ONE} / HADOOP_CONF_DIR_IS_ABS)
+
# Note: can't use -D_FILE_OFFSET_BITS=64, see MAPREDUCE-4258
string(REPLACE "-D_FILE_OFFSET_BITS=64" "" CMAKE_C_FLAGS "${CMAKE_C_FLAGS}")
string(REPLACE "-D_FILE_OFFSET_BITS=64" "" CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS}")
+include(CheckIncludeFiles)
+check_include_files("sys/types.h;sys/sysctl.h" HAVE_SYS_SYSCTL_H)
+
include(CheckFunctionExists)
check_function_exists(canonicalize_file_name HAVE_CANONICALIZE_FILE_NAME)
check_function_exists(fcloseall HAVE_FCLOSEALL)
@@ -32,6 +42,9 @@ check_function_exists(fstatat HAVE_FSTATAT)
check_function_exists(openat HAVE_OPENAT)
check_function_exists(unlinkat HAVE_UNLINKAT)
+include(CheckSymbolExists)
+check_symbol_exists(sysctl "sys/types.h;sys/sysctl.h" HAVE_SYSCTL)
+
if(APPLE)
include_directories( /System/Library/Frameworks )
find_library(COCOA_LIBRARY Cocoa)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9136773/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/config.h.cmake
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/config.h.cmake b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/config.h.cmake
index 0ab7bbd..d8e710f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/config.h.cmake
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/config.h.cmake
@@ -18,14 +18,26 @@
#ifndef CONFIG_H
#define CONFIG_H
+/* custom configs */
+
#cmakedefine HADOOP_CONF_DIR "@HADOOP_CONF_DIR@"
+#cmakedefine HADOOP_CONF_DIR_IS_ABS "@HADOOP_CONF_DIR_IS_ABS@"
+
+/* specific functions */
+
#cmakedefine HAVE_CANONICALIZE_FILE_NAME @HAVE_CANONICALIZE_FILE_NAME@
#cmakedefine HAVE_FCHMODAT @HAVE_FCHMODAT@
#cmakedefine HAVE_FCLOSEALL @HAVE_FCLOSEALL@
#cmakedefine HAVE_FDOPENDIR @HAVE_FDOPENDIR@
#cmakedefine HAVE_FSTATAT @HAVE_FSTATAT@
#cmakedefine HAVE_OPENAT @HAVE_OPENAT@
+#cmakedefine HAVE_SYSCTL @HAVE_SYSCTL@
#cmakedefine HAVE_UNLINKAT @HAVE_UNLINKAT@
+
+/* specific headers */
+
+#cmakedefine HAVE_SYS_SYSCTL_H @HAVE_SYS_SYSCTL_H@
+
#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9136773/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/configuration.h
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/configuration.h b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/configuration.h
index e7e5710..eced13b 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/configuration.h
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/configuration.h
@@ -16,6 +16,10 @@
* limitations under the License.
*/
+#ifdef __FreeBSD__
+#define _WITH_GETLINE
+#endif
+
#include <stddef.h>
/** Define a platform-independent constant instead of using PATH_MAX */
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9136773/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.c
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.c b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.c
index fef7173..e0a97b1 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.c
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.c
@@ -800,7 +800,23 @@ int set_user(const char *user) {
*/
static int change_owner(const char* path, uid_t user, gid_t group) {
if (geteuid() == user && getegid() == group) {
+
+ /*
+ * On the BSDs, this is not a guaranteed shortcut
+ * since group permissions are inherited
+ */
+
+#if defined(__FreeBSD__) || defined(__NetBSD__)
+ if (chown(path, user, group) != 0) {
+ fprintf(LOGFILE, "Can't chown %s to %d:%d - %s\n", path, user, group,
+ strerror(errno));
+ return -1;
+ }
return 0;
+#else
+ return 0;
+#endif
+
} else {
uid_t old_user = geteuid();
gid_t old_group = getegid();
@@ -838,14 +854,14 @@ int create_directory_for_user(const char* path) {
if (ret == 0) {
if (0 == mkdir(path, permissions) || EEXIST == errno) {
// need to reassert the group sticky bit
- if (chmod(path, permissions) != 0) {
- fprintf(LOGFILE, "Can't chmod %s to add the sticky bit - %s\n",
- path, strerror(errno));
- ret = -1;
- } else if (change_owner(path, user, nm_gid) != 0) {
+ if (change_owner(path, user, nm_gid) != 0) {
fprintf(LOGFILE, "Failed to chown %s to %d:%d: %s\n", path, user, nm_gid,
strerror(errno));
ret = -1;
+ } else if (chmod(path, permissions) != 0) {
+ fprintf(LOGFILE, "Can't chmod %s to add the sticky bit - %s\n",
+ path, strerror(errno));
+ ret = -1;
}
} else {
fprintf(LOGFILE, "Failed to create directory %s - %s\n", path,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9136773/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h
index 1bfa3a5..449c7f0 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h
@@ -15,6 +15,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
+/* FreeBSD protects the getline() prototype. See getline(3) for more */
+#ifdef __FreeBSD__
+#define _WITH_GETLINE
+#endif
+
#include <pwd.h>
#include <stdio.h>
#include <sys/types.h>
@@ -96,7 +102,7 @@ extern FILE *LOGFILE;
extern FILE *ERRORFILE;
// get the executable's filename
-char* get_executable();
+char* get_executable(char *argv0);
//function used to load the configurations present in the secure config
void read_executor_config(const char* file_name);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9136773/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/get_executable.c
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/get_executable.c b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/get_executable.c
index cd89735..49ae093 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/get_executable.c
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/get_executable.c
@@ -28,23 +28,27 @@
*
*/
+#include "config.h"
+#include "configuration.h"
+#include "container-executor.h"
+
#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
-#include "config.h"
-#include "configuration.h"
-#include "container-executor.h"
+
+#ifdef HAVE_SYS_SYSCTL_H
+#include <sys/types.h>
+#include <sys/param.h>
+#include <sys/sysctl.h>
+#endif
/*
* A generic function to read a link and return
* the value for use with System V procfs.
* With much thanks to Tom Killian, Roger Faulkner,
* and Ron Gomes, this is pretty generic code.
- *
- * The various BSDs do not have (reliably)
- * have /proc. Custom implementations follow.
*/
char *__get_exec_readproc(char *procfn) {
@@ -53,7 +57,7 @@ char *__get_exec_readproc(char *procfn) {
filename = malloc(EXECUTOR_PATH_MAX);
if (!filename) {
- fprintf(ERRORFILE,"cannot allocate memory for filename: %s\n",strerror(errno));
+ fprintf(ERRORFILE,"cannot allocate memory for filename before readlink: %s\n",strerror(errno));
exit(-1);
}
len = readlink(procfn, filename, EXECUTOR_PATH_MAX);
@@ -62,14 +66,45 @@ char *__get_exec_readproc(char *procfn) {
strerror(errno));
exit(-1);
} else if (len >= EXECUTOR_PATH_MAX) {
- fprintf(ERRORFILE,"Executable name %.*s is longer than %d characters.\n",
- EXECUTOR_PATH_MAX, filename, EXECUTOR_PATH_MAX);
+ fprintf(ERRORFILE,"Resolved path for %s [%s] is longer than %d characters.\n",
+ procfn, filename, EXECUTOR_PATH_MAX);
exit(-1);
}
filename[len] = '\0';
return filename;
}
+
+#ifdef HAVE_SYSCTL
+/*
+ * A generic function to ask the kernel via sysctl.
+ * This is used by most of the open source BSDs, as
+ * many do not reliably have a /proc mounted.
+ */
+
+char *__get_exec_sysctl(int *mib)
+{
+ char buffer[EXECUTOR_PATH_MAX];
+ char *filename;
+ size_t len;
+
+ len = sizeof(buffer);
+ if (sysctl(mib, 4, buffer, &len, NULL, 0) == -1) {
+ fprintf(ERRORFILE,"Can't get executable name from kernel: %s\n",
+ strerror(errno));
+ exit(-1);
+ }
+ filename=malloc(EXECUTOR_PATH_MAX);
+ if (!filename) {
+ fprintf(ERRORFILE,"cannot allocate memory for filename after sysctl: %s\n",strerror(errno));
+ exit(-1);
+ }
+ snprintf(filename,EXECUTOR_PATH_MAX,"%s",buffer);
+ return filename;
+}
+
+#endif /* HAVE_SYSCTL */
+
#ifdef __APPLE__
/*
@@ -80,13 +115,13 @@ char *__get_exec_readproc(char *procfn) {
#include <libproc.h>
-char* get_executable() {
+char* get_executable(char *argv0) {
char *filename;
pid_t pid;
filename = malloc(PROC_PIDPATHINFO_MAXSIZE);
if (!filename) {
- fprintf(ERRORFILE,"cannot allocate memory for filename: %s\n",strerror(errno));
+ fprintf(ERRORFILE,"cannot allocate memory for filename before proc_pidpath: %s\n",strerror(errno));
exit(-1);
}
pid = getpid();
@@ -98,13 +133,33 @@ char* get_executable() {
return filename;
}
+#elif defined(__FreeBSD__)
+
+char* get_executable(char *argv0) {
+ static int mib[] = {
+ CTL_KERN, KERN_PROC, KERN_PROC_PATHNAME, -1
+ };
+ return __get_exec_sysctl(mib);
+}
+
#elif defined(__linux)
-char* get_executable() {
+char* get_executable(char *argv0) {
return __get_exec_readproc("/proc/self/exe");
}
+#elif defined(__NetBSD__) && defined(KERN_PROC_PATHNAME)
+
+/* Only really new NetBSD kernels have KERN_PROC_PATHNAME */
+
+char* get_executable(char *argv0) {
+ static int mib[] = {
+ CTL_KERN, KERN_PROC_ARGS, -1, KERN_PROC_PATHNAME,
+ };
+ return __get_exec_sysctl(mib);
+}
+
#elif defined(__sun)
/*
@@ -116,12 +171,45 @@ char* get_executable() {
* doesn't exist on Solaris hasn't read the proc(4) man page.)
*/
-char* get_executable() {
+char* get_executable(char *argv0) {
return __get_exec_readproc("/proc/self/path/a.out");
}
-#else
+#elif defined(HADOOP_CONF_DIR_IS_ABS)
-#error Cannot safely determine executable path on this operating system.
+/*
+ * This is the fallback for operating systems where
+ * we don't know how to ask the kernel where the executable
+ * is located. It is only used if the maven property
+ * container-executor.conf.dir is set to an absolute path
+ * for security reasons.
+ */
+
+char* get_executable (char *argv0) {
+ char *filename;
+#ifdef HAVE_CANONICALIZE_FILE_NAME
+ filename=canonicalize_file_name(argv0);
+#else
+ filename=realpath(argv0,NULL);
#endif
+
+ if (!filename) {
+ fprintf(ERRORFILE,"realpath of executable: %s\n",strerror(errno));
+ exit(-1);
+ }
+ return filename;
+}
+
+#else
+
+/*
+ * If we ended up here, we're on an operating system that doesn't
+ * match any of the above. This means either the OS needs to get a
+ * code added or the container-executor.conf.dir maven property
+ * should be set to an absolute path.
+ */
+
+#error Cannot safely determine executable path with a relative HADOOP_CONF_DIR on this operating system.
+
+#endif /* platform checks */
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9136773/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/main.c
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/main.c b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/main.c
index 6fade5c..5fc92d6 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/main.c
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/main.c
@@ -96,9 +96,9 @@ in case of validation failures. Also sets up configuration / group information e
This function is to be called in every invocation of container-executor, irrespective
of whether an explicit checksetup operation is requested. */
-static void assert_valid_setup() {
+static void assert_valid_setup(char *argv0) {
int ret;
- char *executable_file = get_executable();
+ char *executable_file = get_executable(argv0);
if (!executable_file) {
fprintf(ERRORFILE,"realpath of executable: %s\n",strerror(errno));
flush_and_close_log_files();
@@ -429,7 +429,7 @@ static int validate_run_as_user_commands(int argc, char **argv, int *operation)
int main(int argc, char **argv) {
open_log_files();
- assert_valid_setup();
+ assert_valid_setup(argv[0]);
int operation;
int ret = validate_arguments(argc, argv, &operation);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9136773/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/test/test-container-executor.c
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/test/test-container-executor.c b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/test/test-container-executor.c
index a2dead2..5166097 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/test/test-container-executor.c
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/test/test-container-executor.c
@@ -826,6 +826,7 @@ int main(int argc, char **argv) {
LOGFILE = stdout;
ERRORFILE = stderr;
+ printf("Attempting to clean up from any previous runs\n");
// clean up any junk from previous run
if (system("chmod -R u=rwx " TEST_ROOT "; rm -fr " TEST_ROOT)) {
exit(1);
@@ -838,6 +839,9 @@ int main(int argc, char **argv) {
if (write_config_file(TEST_ROOT "/test.cfg", 1) != 0) {
exit(1);
}
+
+ printf("\nOur executable is %s\n",get_executable(argv[0]));
+
read_executor_config(TEST_ROOT "/test.cfg");
local_dirs = extract_values(strdup(NM_LOCAL_DIRS));
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[26/30] hadoop git commit: HADOOP-13458.
LoadBalancingKMSClientProvider#doOp should log IOException stacktrace.
Contributed by Chen Liang.
Posted by dr...@apache.org.
HADOOP-13458. LoadBalancingKMSClientProvider#doOp should log IOException stacktrace. Contributed by Chen Liang.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/a1f6564d
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/a1f6564d
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/a1f6564d
Branch: refs/heads/HADOOP-12756
Commit: a1f6564d31c16582ea949e4448091f1cc56c7dfe
Parents: 6f63566
Author: Wei-Chiu Chuang <we...@apache.org>
Authored: Wed Aug 3 14:19:08 2016 -0700
Committer: Wei-Chiu Chuang <we...@apache.org>
Committed: Wed Aug 3 14:19:47 2016 -0700
----------------------------------------------------------------------
.../hadoop/crypto/key/kms/LoadBalancingKMSClientProvider.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a1f6564d/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/kms/LoadBalancingKMSClientProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/kms/LoadBalancingKMSClientProvider.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/kms/LoadBalancingKMSClientProvider.java
index 6a9bd62..e6ff079 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/kms/LoadBalancingKMSClientProvider.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/kms/LoadBalancingKMSClientProvider.java
@@ -33,6 +33,7 @@ import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension.EncryptedKeyVersi
import org.apache.hadoop.crypto.key.KeyProviderDelegationTokenExtension;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -93,8 +94,8 @@ public class LoadBalancingKMSClientProvider extends KeyProvider implements
try {
return op.call(provider);
} catch (IOException ioe) {
- LOG.warn("KMS provider at [{}] threw an IOException [{}]!!",
- provider.getKMSUrl(), ioe.getMessage());
+ LOG.warn("KMS provider at [{}] threw an IOException!! {}",
+ provider.getKMSUrl(), StringUtils.stringifyException(ioe));
ex = ioe;
} catch (Exception e) {
if (e instanceof RuntimeException) {
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[22/30] hadoop git commit: HDFS-10710. In
BlockManager#rescanPostponedMisreplicatedBlocks(),
postponed misreplicated block counts should be retrieved within the NN lock
protection.Contributed by GAO Rui.
Posted by dr...@apache.org.
HDFS-10710. In BlockManager#rescanPostponedMisreplicatedBlocks(), postponed misreplicated block counts should be retrieved within the NN lock protection.Contributed by GAO Rui.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/f4ba5ff1
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/f4ba5ff1
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/f4ba5ff1
Branch: refs/heads/HADOOP-12756
Commit: f4ba5ff1d70ef92d59851c09c4bd4b43d6c04971
Parents: db64654
Author: Jing Zhao <ji...@apache.org>
Authored: Wed Aug 3 11:39:06 2016 -0700
Committer: Jing Zhao <ji...@apache.org>
Committed: Wed Aug 3 12:42:05 2016 -0700
----------------------------------------------------------------------
.../apache/hadoop/hdfs/server/blockmanagement/BlockManager.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4ba5ff1/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
index 349b018..6c25d64 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
@@ -2278,9 +2278,9 @@ public class BlockManager implements BlockStatsMXBean {
return;
}
long startTimeRescanPostponedMisReplicatedBlocks = Time.monotonicNow();
+ namesystem.writeLock();
long startPostponedMisReplicatedBlocksCount =
getPostponedMisreplicatedBlocksCount();
- namesystem.writeLock();
try {
// blocksPerRescan is the configured number of blocks per rescan.
// Randomly select blocksPerRescan consecutive blocks from the HashSet
@@ -2333,9 +2333,9 @@ public class BlockManager implements BlockStatsMXBean {
}
}
} finally {
- namesystem.writeUnlock();
long endPostponedMisReplicatedBlocksCount =
getPostponedMisreplicatedBlocksCount();
+ namesystem.writeUnlock();
LOG.info("Rescan of postponedMisreplicatedBlocks completed in " +
(Time.monotonicNow() - startTimeRescanPostponedMisReplicatedBlocks) +
" msecs. " + endPostponedMisReplicatedBlocksCount +
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[05/30] hadoop git commit: YARN-5458. Rename DockerStopCommandTest to
TestDockerStopCommand. Contributed by Shane Kumpf.
Posted by dr...@apache.org.
YARN-5458. Rename DockerStopCommandTest to TestDockerStopCommand. Contributed by Shane Kumpf.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/7fc70c64
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/7fc70c64
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/7fc70c64
Branch: refs/heads/HADOOP-12756
Commit: 7fc70c6422da3602ad9d4364493f25454a1de50c
Parents: 2e7c2a1
Author: Varun Vasudev <vv...@apache.org>
Authored: Tue Aug 2 16:56:45 2016 +0530
Committer: Varun Vasudev <vv...@apache.org>
Committed: Tue Aug 2 16:56:45 2016 +0530
----------------------------------------------------------------------
.../runtime/docker/DockerStopCommandTest.java | 55 --------------------
.../runtime/docker/TestDockerStopCommand.java | 55 ++++++++++++++++++++
2 files changed, 55 insertions(+), 55 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7fc70c64/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerStopCommandTest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerStopCommandTest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerStopCommandTest.java
deleted file mode 100644
index b53e58b..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerStopCommandTest.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * /
- */
-
-package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker;
-
-import static org.junit.Assert.*;
-import org.junit.Before;
-import org.junit.Test;
-
-/**
- * Tests the docker stop command and its command
- * line arguments.
- */
-public class DockerStopCommandTest {
-
- private DockerStopCommand dockerStopCommand;
-
- private static final int GRACE_PERIOD = 10;
- private static final String CONTAINER_NAME = "foo";
-
- @Before
- public void setup() {
- dockerStopCommand = new DockerStopCommand(CONTAINER_NAME);
- }
-
- @Test
- public void testGetCommandOption() {
- assertEquals("stop", dockerStopCommand.getCommandOption());
- }
-
- @Test
- public void testSetGracePeriod() throws Exception {
- dockerStopCommand.setGracePeriod(GRACE_PERIOD);
- assertEquals("stop foo --time=10",
- dockerStopCommand.getCommandWithArguments());
-
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7fc70c64/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerStopCommand.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerStopCommand.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerStopCommand.java
new file mode 100644
index 0000000..c9743f3
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerStopCommand.java
@@ -0,0 +1,55 @@
+/*
+ * *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * /
+ */
+
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests the docker stop command and its command
+ * line arguments.
+ */
+public class TestDockerStopCommand {
+
+ private DockerStopCommand dockerStopCommand;
+
+ private static final int GRACE_PERIOD = 10;
+ private static final String CONTAINER_NAME = "foo";
+
+ @Before
+ public void setup() {
+ dockerStopCommand = new DockerStopCommand(CONTAINER_NAME);
+ }
+
+ @Test
+ public void testGetCommandOption() {
+ assertEquals("stop", dockerStopCommand.getCommandOption());
+ }
+
+ @Test
+ public void testSetGracePeriod() throws Exception {
+ dockerStopCommand.setGracePeriod(GRACE_PERIOD);
+ assertEquals("stop foo --time=10",
+ dockerStopCommand.getCommandWithArguments());
+
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[24/30] hadoop git commit: HADOOP-10980. TestActiveStandbyElector
fails occasionally in trunk. Contributed by Eric Badger
Posted by dr...@apache.org.
HADOOP-10980. TestActiveStandbyElector fails occasionally in trunk. Contributed by Eric Badger
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/c8274543
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/c8274543
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/c8274543
Branch: refs/heads/HADOOP-12756
Commit: c82745432a962c817a8a7db92bb830fb6af01e33
Parents: db4a61d
Author: Jason Lowe <jl...@apache.org>
Authored: Wed Aug 3 20:17:30 2016 +0000
Committer: Jason Lowe <jl...@apache.org>
Committed: Wed Aug 3 20:17:30 2016 +0000
----------------------------------------------------------------------
.../apache/hadoop/ha/ActiveStandbyElector.java | 19 +++++++++++++++----
.../hadoop/ha/TestActiveStandbyElector.java | 12 ++++++++++--
2 files changed, 25 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c8274543/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ActiveStandbyElector.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ActiveStandbyElector.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ActiveStandbyElector.java
index cb2e081..9a90ccf 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ActiveStandbyElector.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ActiveStandbyElector.java
@@ -667,13 +667,13 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
/**
* Get a new zookeeper client instance. protected so that test class can
- * inherit and pass in a mock object for zookeeper
+ * inherit and mock out the zookeeper instance
*
* @return new zookeeper client instance
* @throws IOException
* @throws KeeperException zookeeper connectionloss exception
*/
- protected synchronized ZooKeeper getNewZooKeeper() throws IOException,
+ protected synchronized ZooKeeper connectToZooKeeper() throws IOException,
KeeperException {
// Unfortunately, the ZooKeeper constructor connects to ZooKeeper and
@@ -682,7 +682,7 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
// we construct the watcher first, and have it block any events it receives
// before we can set its ZooKeeper reference.
watcher = new WatcherWithClientRef();
- ZooKeeper zk = new ZooKeeper(zkHostPort, zkSessionTimeout, watcher);
+ ZooKeeper zk = createZooKeeper();
watcher.setZooKeeperRef(zk);
// Wait for the asynchronous success/failure. This may throw an exception
@@ -695,6 +695,17 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
return zk;
}
+ /**
+ * Get a new zookeeper client instance. protected so that test class can
+ * inherit and pass in a mock object for zookeeper
+ *
+ * @return new zookeeper client instance
+ * @throws IOException
+ */
+ protected ZooKeeper createZooKeeper() throws IOException {
+ return new ZooKeeper(zkHostPort, zkSessionTimeout, watcher);
+ }
+
private void fatalError(String errorMessage) {
LOG.fatal(errorMessage);
reset();
@@ -830,7 +841,7 @@ public class ActiveStandbyElector implements StatCallback, StringCallback {
zkClient = null;
watcher = null;
}
- zkClient = getNewZooKeeper();
+ zkClient = connectToZooKeeper();
if (LOG.isDebugEnabled()) {
LOG.debug("Created new connection for " + this);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c8274543/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java
index 83a3a4f..8c6f7a3 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestActiveStandbyElector.java
@@ -66,7 +66,7 @@ public class TestActiveStandbyElector {
}
@Override
- public ZooKeeper getNewZooKeeper() {
+ public ZooKeeper connectToZooKeeper() {
++count;
return mockZK;
}
@@ -749,7 +749,15 @@ public class TestActiveStandbyElector {
try {
new ActiveStandbyElector("127.0.0.1", 2000, ZK_PARENT_NAME,
Ids.OPEN_ACL_UNSAFE, Collections.<ZKAuthInfo> emptyList(), mockApp,
- CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT);
+ CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT) {
+
+ @Override
+ protected ZooKeeper createZooKeeper() throws IOException {
+ return Mockito.mock(ZooKeeper.class);
+ }
+ };
+
+
Assert.fail("Did not throw zookeeper connection loss exceptions!");
} catch (KeeperException ke) {
GenericTestUtils.assertExceptionContains( "ConnectionLoss", ke);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[28/30] hadoop git commit: HADOOP-13443. KMS should check the type of
underlying keyprovider of KeyProviderExtension before falling back to
default. Contributed by Anthony Young-Garner.
Posted by dr...@apache.org.
HADOOP-13443. KMS should check the type of underlying keyprovider of KeyProviderExtension before falling back to default. Contributed by Anthony Young-Garner.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/05db6491
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/05db6491
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/05db6491
Branch: refs/heads/HADOOP-12756
Commit: 05db64913d8c620cf481f74219bac9b72cb142af
Parents: 8f1c374
Author: Xiao Chen <xi...@apache.org>
Authored: Thu Aug 4 15:54:52 2016 +0800
Committer: Xiao Chen <xi...@apache.org>
Committed: Thu Aug 4 15:57:34 2016 +0800
----------------------------------------------------------------------
.../crypto/key/KeyProviderCryptoExtension.java | 31 ++-
.../key/TestKeyProviderCryptoExtension.java | 208 +++++++++++++++++++
2 files changed, 231 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/05db6491/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProviderCryptoExtension.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProviderCryptoExtension.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProviderCryptoExtension.java
index 7e95211..1ecd9f6 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProviderCryptoExtension.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProviderCryptoExtension.java
@@ -389,24 +389,39 @@ public class KeyProviderCryptoExtension extends
}
/**
- * Creates a <code>KeyProviderCryptoExtension</code> using a given
+ * Creates a <code>KeyProviderCryptoExtension</code> using a given
* {@link KeyProvider}.
* <p/>
- * If the given <code>KeyProvider</code> implements the
+ * If the given <code>KeyProvider</code> implements the
* {@link CryptoExtension} interface the <code>KeyProvider</code> itself
- * will provide the extension functionality, otherwise a default extension
+ * will provide the extension functionality.
+ * If the given <code>KeyProvider</code> implements the
+ * {@link KeyProviderExtension} interface and the KeyProvider being
+ * extended by the <code>KeyProvider</code> implements the
+ * {@link CryptoExtension} interface, the KeyProvider being extended will
+ * provide the extension functionality. Otherwise, a default extension
* implementation will be used.
- *
- * @param keyProvider <code>KeyProvider</code> to use to create the
+ *
+ * @param keyProvider <code>KeyProvider</code> to use to create the
* <code>KeyProviderCryptoExtension</code> extension.
* @return a <code>KeyProviderCryptoExtension</code> instance using the
* given <code>KeyProvider</code>.
*/
public static KeyProviderCryptoExtension createKeyProviderCryptoExtension(
KeyProvider keyProvider) {
- CryptoExtension cryptoExtension = (keyProvider instanceof CryptoExtension)
- ? (CryptoExtension) keyProvider
- : new DefaultCryptoExtension(keyProvider);
+ CryptoExtension cryptoExtension = null;
+ if (keyProvider instanceof CryptoExtension) {
+ cryptoExtension = (CryptoExtension) keyProvider;
+ } else if (keyProvider instanceof KeyProviderExtension &&
+ ((KeyProviderExtension)keyProvider).getKeyProvider() instanceof
+ KeyProviderCryptoExtension.CryptoExtension) {
+ KeyProviderExtension keyProviderExtension =
+ (KeyProviderExtension)keyProvider;
+ cryptoExtension =
+ (CryptoExtension)keyProviderExtension.getKeyProvider();
+ } else {
+ cryptoExtension = new DefaultCryptoExtension(keyProvider);
+ }
return new KeyProviderCryptoExtension(keyProvider, cryptoExtension);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/05db6491/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/key/TestKeyProviderCryptoExtension.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/key/TestKeyProviderCryptoExtension.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/key/TestKeyProviderCryptoExtension.java
index 62e3310..4712698 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/key/TestKeyProviderCryptoExtension.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/key/TestKeyProviderCryptoExtension.java
@@ -17,9 +17,13 @@
*/
package org.apache.hadoop.crypto.key;
+import java.io.IOException;
import java.net.URI;
+import java.net.URISyntaxException;
+import java.security.GeneralSecurityException;
import java.security.SecureRandom;
import java.util.Arrays;
+import java.util.List;
import javax.crypto.Cipher;
import javax.crypto.spec.IvParameterSpec;
@@ -27,6 +31,7 @@ import javax.crypto.spec.SecretKeySpec;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension.EncryptedKeyVersion;
+import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -132,4 +137,207 @@ public class TestKeyProviderCryptoExtension {
assertArrayEquals("Wrong key material from decryptEncryptedKey",
manualMaterial, apiMaterial);
}
+
+ @Test
+ public void testNonDefaultCryptoExtensionSelectionWithCachingKeyProvider()
+ throws Exception {
+ Configuration config = new Configuration();
+ KeyProvider localKp = new DummyCryptoExtensionKeyProvider(config);
+ localKp = new CachingKeyProvider(localKp, 30000, 30000);
+ EncryptedKeyVersion localEkv = getEncryptedKeyVersion(config, localKp);
+ Assert.assertEquals("dummyFakeKey@1",
+ localEkv.getEncryptionKeyVersionName());
+ }
+
+ @Test
+ public void testDefaultCryptoExtensionSelectionWithCachingKeyProvider()
+ throws Exception {
+ Configuration config = new Configuration();
+ KeyProvider localKp =
+ new UserProvider.Factory().
+ createProvider(new URI("user:///"), config);
+ localKp = new CachingKeyProvider(localKp, 30000, 30000);
+ EncryptedKeyVersion localEkv = getEncryptedKeyVersion(config, localKp);
+ Assert.assertEquals(ENCRYPTION_KEY_NAME+"@0",
+ localEkv.getEncryptionKeyVersionName());
+ }
+
+ @Test
+ public void testNonDefaultCryptoExtensionSelectionOnKeyProviderExtension()
+ throws Exception {
+ Configuration config = new Configuration();
+ KeyProvider localKp = new UserProvider.Factory().
+ createProvider(new URI("user:///"), config);
+ localKp = new DummyCachingCryptoExtensionKeyProvider(localKp, 30000, 30000);
+ EncryptedKeyVersion localEkv = getEncryptedKeyVersion(config, localKp);
+ Assert.assertEquals("dummyCachingFakeKey@1",
+ localEkv.getEncryptionKeyVersionName());
+ }
+
+ private EncryptedKeyVersion getEncryptedKeyVersion(Configuration config,
+ KeyProvider localKp)
+ throws IOException, GeneralSecurityException {
+ KeyProvider.Options localOptions = new KeyProvider.Options(config);
+ localOptions.setCipher(CIPHER);
+ localOptions.setBitLength(128);
+ KeyVersion localEncryptionKey =
+ localKp.createKey(ENCRYPTION_KEY_NAME,
+ SecureRandom.getSeed(16), localOptions);
+ KeyProviderCryptoExtension localKpExt =
+ KeyProviderCryptoExtension.
+ createKeyProviderCryptoExtension(localKp);
+ return localKpExt.generateEncryptedKey(localEncryptionKey.getName());
+ }
+
+ /**
+ * Dummy class to test that this key provider is chosen to
+ * provide CryptoExtension services over the DefaultCryptoExtension.
+ */
+ public class DummyCryptoExtensionKeyProvider extends KeyProvider
+ implements KeyProviderCryptoExtension.CryptoExtension {
+
+ private KeyProvider kp;
+ private KeyVersion kv;
+ private EncryptedKeyVersion ekv;
+
+ public DummyCryptoExtensionKeyProvider(Configuration conf) {
+ super(conf);
+ conf = new Configuration();
+ try {
+ this.kp = new UserProvider.Factory().createProvider(
+ new URI("user:///"), conf);
+ this.kv = new KeyVersion(ENCRYPTION_KEY_NAME,
+ "dummyFakeKey@1", new byte[16]);
+ this.ekv = new EncryptedKeyVersion(ENCRYPTION_KEY_NAME,
+ "dummyFakeKey@1", new byte[16], kv);
+ } catch (URISyntaxException e) {
+ fail(e.getMessage());
+ } catch (IOException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Override
+ public void warmUpEncryptedKeys(String... keyNames) throws IOException {
+
+ }
+
+ @Override
+ public void drain(String keyName) {
+
+ }
+
+ @Override
+ public EncryptedKeyVersion generateEncryptedKey(String encryptionKeyName)
+ throws IOException, GeneralSecurityException {
+ return this.ekv;
+ }
+
+ @Override
+ public KeyVersion decryptEncryptedKey(
+ EncryptedKeyVersion encryptedKeyVersion)
+ throws IOException, GeneralSecurityException {
+ return kv;
+ }
+
+ @Override
+ public KeyVersion getKeyVersion(String versionName)
+ throws IOException {
+ return this.kp.getKeyVersion(versionName);
+ }
+
+ @Override
+ public List<String> getKeys() throws IOException {
+ return this.kp.getKeys();
+ }
+
+ @Override
+ public List<KeyVersion> getKeyVersions(String name)
+ throws IOException {
+ return this.kp.getKeyVersions(name);
+ }
+
+ @Override
+ public Metadata getMetadata(String name)
+ throws IOException {
+ return this.kp.getMetadata(name);
+ }
+
+ @Override
+ public KeyVersion createKey(String name, byte[] material,
+ Options localOptions) throws IOException {
+ return this.kp.createKey(name, material, localOptions);
+ }
+
+ @Override
+ public void deleteKey(String name) throws IOException {
+ this.kp.deleteKey(name);
+ }
+
+ @Override
+ public KeyVersion rollNewVersion(String name,
+ byte[] material) throws IOException {
+ return this.kp.rollNewVersion(name, material);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ this.kp.flush();
+ }
+ }
+
+ /**
+ * Dummy class to verify that CachingKeyProvider is used to
+ * provide CryptoExtension services if the CachingKeyProvider itself
+ * implements CryptoExtension.
+ */
+ public class DummyCachingCryptoExtensionKeyProvider
+ extends CachingKeyProvider
+ implements KeyProviderCryptoExtension.CryptoExtension {
+ private KeyProvider kp;
+ private KeyVersion kv;
+ private EncryptedKeyVersion ekv;
+
+ public DummyCachingCryptoExtensionKeyProvider(KeyProvider keyProvider,
+ long keyTimeoutMillis,
+ long currKeyTimeoutMillis) {
+ super(keyProvider, keyTimeoutMillis, currKeyTimeoutMillis);
+ conf = new Configuration();
+ try {
+ this.kp = new UserProvider.Factory().createProvider(
+ new URI("user:///"), conf);
+ this.kv = new KeyVersion(ENCRYPTION_KEY_NAME,
+ "dummyCachingFakeKey@1", new byte[16]);
+ this.ekv = new EncryptedKeyVersion(ENCRYPTION_KEY_NAME,
+ "dummyCachingFakeKey@1", new byte[16], kv);
+ } catch (URISyntaxException e) {
+ fail(e.getMessage());
+ } catch (IOException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Override
+ public void warmUpEncryptedKeys(String... keyNames) throws IOException {
+
+ }
+
+ @Override
+ public void drain(String keyName) {
+
+ }
+
+ @Override
+ public EncryptedKeyVersion generateEncryptedKey(String encryptionKeyName)
+ throws IOException, GeneralSecurityException {
+ return this.ekv;
+ }
+
+ @Override
+ public KeyVersion decryptEncryptedKey(
+ EncryptedKeyVersion encryptedKeyVersion)
+ throws IOException, GeneralSecurityException {
+ return kv;
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[25/30] hadoop git commit: HDFS-10569. A bug causes OutOfIndex error
in BlockListAsLongs. Contributed by Weiwei Yang.
Posted by dr...@apache.org.
HDFS-10569. A bug causes OutOfIndex error in BlockListAsLongs. Contributed by Weiwei Yang.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/6f635666
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/6f635666
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/6f635666
Branch: refs/heads/HADOOP-12756
Commit: 6f63566694f8cec64a469448a8fa00ce921ce367
Parents: c827454
Author: Kihwal Lee <ki...@apache.org>
Authored: Wed Aug 3 15:19:31 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Wed Aug 3 15:20:20 2016 -0500
----------------------------------------------------------------------
.../hadoop/hdfs/protocol/BlockListAsLongs.java | 4 ++--
.../hdfs/protocol/TestBlockListAsLongs.java | 19 ++++++++++++++++++-
2 files changed, 20 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/6f635666/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
index 26c7ffb..8f482e3 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
@@ -468,8 +468,8 @@ public abstract class BlockListAsLongs implements Iterable<BlockReportReplica> {
long[] longs = new long[2+values.size()];
longs[0] = finalizedBlocks;
longs[1] = numBlocks - finalizedBlocks;
- for (int i=0; i < longs.length; i++) {
- longs[i] = values.get(i);
+ for(int i=0; i<values.size(); i++) {
+ longs[2+i] = values.get(i);
}
return longs;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/6f635666/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestBlockListAsLongs.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestBlockListAsLongs.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestBlockListAsLongs.java
index 2cc1f7d..cd2b5e8 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestBlockListAsLongs.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestBlockListAsLongs.java
@@ -188,6 +188,20 @@ public class TestBlockListAsLongs {
assertTrue(reportReplicas.isEmpty());
}
+ private BlockListAsLongs getBlockList(Replica ... replicas) {
+ int numBlocks = replicas.length;
+ List<Long> longs = new ArrayList<Long>(2 + numBlocks);
+ longs.add(Long.valueOf(numBlocks));
+ longs.add(0L);
+ for(Replica r : replicas) {
+ longs.add(r.getBlockId());
+ longs.add(r.getBytesOnDisk());
+ longs.add(r.getGenerationStamp());
+ }
+ BlockListAsLongs blockList = BlockListAsLongs.decodeLongs(longs);
+ return blockList;
+ }
+
@Test
public void testCapabilitiesInited() {
NamespaceInfo nsInfo = new NamespaceInfo();
@@ -237,7 +251,10 @@ public class TestBlockListAsLongs {
// back up to prior version and check DN sends old-style BR
request.set(null);
nsInfo.setCapabilities(Capability.UNKNOWN.getMask());
- nn.blockReport(reg, "pool", sbr,
+ BlockListAsLongs blockList = getBlockList(r);
+ StorageBlockReport[] obp = new StorageBlockReport[] {
+ new StorageBlockReport(new DatanodeStorage("s1"), blockList) };
+ nn.blockReport(reg, "pool", obp,
new BlockReportContext(1, 0, System.nanoTime(), 0L, true));
proto = request.get();
assertNotNull(proto);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[21/30] hadoop git commit: YARN-5462.
TestNodeStatusUpdater.testNodeStatusUpdaterRetryAndNMShutdown fails
intermittently. Contributed by Eric Badger
Posted by dr...@apache.org.
YARN-5462. TestNodeStatusUpdater.testNodeStatusUpdaterRetryAndNMShutdown fails intermittently. Contributed by Eric Badger
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/db646540
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/db646540
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/db646540
Branch: refs/heads/HADOOP-12756
Commit: db646540f094077941b56ed681a4f3e5853f5b7f
Parents: 4d92aef
Author: Jason Lowe <jl...@apache.org>
Authored: Wed Aug 3 19:17:25 2016 +0000
Committer: Jason Lowe <jl...@apache.org>
Committed: Wed Aug 3 19:17:25 2016 +0000
----------------------------------------------------------------------
.../hadoop/yarn/server/nodemanager/TestNodeStatusUpdater.java | 5 +++++
1 file changed, 5 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/db646540/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeStatusUpdater.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeStatusUpdater.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeStatusUpdater.java
index e7d18b7..c71b1e6 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeStatusUpdater.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeStatusUpdater.java
@@ -561,6 +561,8 @@ public class TestNodeStatusUpdater {
@Override
protected void serviceStop() throws Exception {
+ // Make sure that all containers are started before starting shutdown
+ syncBarrier.await(10000, TimeUnit.MILLISECONDS);
System.out.println("Called stooppppp");
super.serviceStop();
isStopped = true;
@@ -1757,6 +1759,9 @@ public class TestNodeStatusUpdater {
new File("start_file.txt"), port);
try {
+ // Wait until we start stopping
+ syncBarrier.await(10000, TimeUnit.MILLISECONDS);
+ // Wait until we finish stopping
syncBarrier.await(10000, TimeUnit.MILLISECONDS);
} catch (Exception e) {
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[10/30] hadoop git commit: HADOOP-13434. Add bash quoting to Shell
class. (Owen O'Malley)
Posted by dr...@apache.org.
HADOOP-13434. Add bash quoting to Shell class. (Owen O'Malley)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/954465e7
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/954465e7
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/954465e7
Branch: refs/heads/HADOOP-12756
Commit: 954465e7ba3af3e5b083b6251562e5e77529f906
Parents: 5e5b879
Author: Arpit Agarwal <ar...@apache.org>
Authored: Tue Aug 2 13:40:33 2016 -0700
Committer: Arpit Agarwal <ar...@apache.org>
Committed: Tue Aug 2 13:40:33 2016 -0700
----------------------------------------------------------------------
.../main/java/org/apache/hadoop/util/Shell.java | 103 ++++++++++++-------
.../java/org/apache/hadoop/util/TestShell.java | 14 ++-
2 files changed, 73 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/954465e7/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/Shell.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/Shell.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/Shell.java
index ea8db07..60f8ff5 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/Shell.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/Shell.java
@@ -42,7 +42,7 @@ import org.slf4j.LoggerFactory;
* A base class for running a Shell command.
*
* <code>Shell</code> can be used to run shell commands like <code>du</code> or
- * <code>df</code>. It also offers facilities to gate commands by
+ * <code>df</code>. It also offers facilities to gate commands by
* time-intervals.
*/
@InterfaceAudience.Public
@@ -118,6 +118,21 @@ public abstract class Shell {
}
}
+ /**
+ * Quote the given arg so that bash will interpret it as a single value.
+ * Note that this quotes it for one level of bash, if you are passing it
+ * into a badly written shell script, you need to fix your shell script.
+ * @param arg the argument to quote
+ * @return the quoted string
+ */
+ static String bashQuote(String arg) {
+ StringBuilder buffer = new StringBuilder(arg.length() + 2);
+ buffer.append('\'');
+ buffer.append(arg.replace("'", "'\\''"));
+ buffer.append('\'');
+ return buffer.toString();
+ }
+
/** a Unix command to get the current user's name: {@value}. */
public static final String USER_NAME_COMMAND = "whoami";
@@ -173,7 +188,7 @@ public abstract class Shell {
/** a Unix command to get the current user's groups list. */
public static String[] getGroupsCommand() {
return (WINDOWS)? new String[]{"cmd", "/c", "groups"}
- : new String[]{"bash", "-c", "groups"};
+ : new String[]{"groups"};
}
/**
@@ -184,10 +199,14 @@ public abstract class Shell {
*/
public static String[] getGroupsForUserCommand(final String user) {
//'groups username' command return is inconsistent across different unixes
- return WINDOWS ?
- new String[]
- {getWinUtilsPath(), "groups", "-F", "\"" + user + "\""}
- : new String[] {"bash", "-c", "id -gn " + user + "; id -Gn " + user};
+ if (WINDOWS) {
+ return new String[]
+ {getWinUtilsPath(), "groups", "-F", "\"" + user + "\""};
+ } else {
+ String quotedUser = bashQuote(user);
+ return new String[] {"bash", "-c", "id -gn " + quotedUser +
+ "; id -Gn " + quotedUser};
+ }
}
/**
@@ -199,17 +218,20 @@ public abstract class Shell {
*/
public static String[] getGroupsIDForUserCommand(final String user) {
//'groups username' command return is inconsistent across different unixes
- return WINDOWS ?
- new String[]
- {getWinUtilsPath(), "groups", "-F", "\"" + user + "\""}
- : new String[] {"bash", "-c", "id -g " + user + "; id -G " + user};
+ if (WINDOWS) {
+ return new String[]{getWinUtilsPath(), "groups", "-F", "\"" + user +
+ "\""};
+ } else {
+ String quotedUser = bashQuote(user);
+ return new String[] {"bash", "-c", "id -g " + quotedUser + "; id -G " +
+ quotedUser};
+ }
}
/** A command to get a given netgroup's user list. */
public static String[] getUsersForNetgroupCommand(final String netgroup) {
//'groups username' command return is non-consistent across different unixes
- return WINDOWS ? new String [] {"cmd", "/c", "getent netgroup " + netgroup}
- : new String [] {"bash", "-c", "getent netgroup " + netgroup};
+ return new String[] {"getent", "netgroup", netgroup};
}
/** Return a command to get permission information. */
@@ -233,14 +255,15 @@ public abstract class Shell {
/**
* Return a command to set permission for specific file.
- *
+ *
* @param perm String permission to set
* @param recursive boolean true to apply to all sub-directories recursively
* @param file String file to set
* @return String[] containing command and arguments
*/
public static String[] getSetPermissionCommand(String perm,
- boolean recursive, String file) {
+ boolean recursive,
+ String file) {
String[] baseCmd = getSetPermissionCommand(perm, recursive);
String[] cmdWithFile = Arrays.copyOf(baseCmd, baseCmd.length + 1);
cmdWithFile[cmdWithFile.length - 1] = file;
@@ -290,9 +313,9 @@ public abstract class Shell {
if (isSetsidAvailable) {
// Use the shell-builtin as it support "--" in all Hadoop supported OSes
- return new String[] { "bash", "-c", "kill -" + code + " -- -" + pid };
+ return new String[] {"kill", "-" + code, "--", "-" + pid};
} else {
- return new String[] { "bash", "-c", "kill -" + code + " " + pid };
+ return new String[] {"kill", "-" + code, pid };
}
}
@@ -310,7 +333,7 @@ public abstract class Shell {
* Returns a File referencing a script with the given basename, inside the
* given parent directory. The file extension is inferred by platform:
* <code>".cmd"</code> on Windows, or <code>".sh"</code> otherwise.
- *
+ *
* @param parent File parent directory
* @param basename String script file basename
* @return File referencing the script in the directory
@@ -342,8 +365,8 @@ public abstract class Shell {
public static String[] getRunScriptCommand(File script) {
String absolutePath = script.getAbsolutePath();
return WINDOWS ?
- new String[] { "cmd", "/c", absolutePath }
- : new String[] { "/bin/bash", absolutePath };
+ new String[] {"cmd", "/c", absolutePath }
+ : new String[] {"/bin/bash", bashQuote(absolutePath) };
}
/** a Unix command to set permission: {@value}. */
@@ -527,11 +550,11 @@ public abstract class Shell {
/**
* Fully qualify the path to a binary that should be in a known hadoop
- * bin location. This is primarily useful for disambiguating call-outs
- * to executable sub-components of Hadoop to avoid clashes with other
- * executables that may be in the path. Caveat: this call doesn't
- * just format the path to the bin directory. It also checks for file
- * existence of the composed path. The output of this call should be
+ * bin location. This is primarily useful for disambiguating call-outs
+ * to executable sub-components of Hadoop to avoid clashes with other
+ * executables that may be in the path. Caveat: this call doesn't
+ * just format the path to the bin directory. It also checks for file
+ * existence of the composed path. The output of this call should be
* cached by callers.
*
* @param executable executable
@@ -851,7 +874,7 @@ public abstract class Shell {
}
/** Run the command. */
- private void runCommand() throws IOException {
+ private void runCommand() throws IOException {
ProcessBuilder builder = new ProcessBuilder(getExecString());
Timer timeOutTimer = null;
ShellTimeoutTimerTask timeoutTimerTask = null;
@@ -873,7 +896,7 @@ public abstract class Shell {
}
builder.redirectErrorStream(redirectErrorStream);
-
+
if (Shell.WINDOWS) {
synchronized (WindowsProcessLaunchLock) {
// To workaround the race condition issue with child processes
@@ -894,7 +917,7 @@ public abstract class Shell {
//One time scheduling.
timeOutTimer.schedule(timeoutTimerTask, timeOutInterval);
}
- final BufferedReader errReader =
+ final BufferedReader errReader =
new BufferedReader(new InputStreamReader(
process.getErrorStream(), Charset.defaultCharset()));
BufferedReader inReader =
@@ -932,7 +955,7 @@ public abstract class Shell {
parseExecResult(inReader); // parse the output
// clear the input stream buffer
String line = inReader.readLine();
- while(line != null) {
+ while(line != null) {
line = inReader.readLine();
}
// wait for the process to finish and check the exit code
@@ -1069,13 +1092,13 @@ public abstract class Shell {
/**
* A simple shell command executor.
- *
+ *
* <code>ShellCommandExecutor</code>should be used in cases where the output
* of the command needs no explicit parsing and where the command, working
* directory and the environment remains unchanged. The output of the command
* is stored as-is and is expected to be small.
*/
- public static class ShellCommandExecutor extends Shell
+ public static class ShellCommandExecutor extends Shell
implements CommandExecutor {
private String[] command;
@@ -1102,7 +1125,7 @@ public abstract class Shell {
/**
* Create a new instance of the ShellCommandExecutor to execute a command.
- *
+ *
* @param execString The command to execute with arguments
* @param dir If not-null, specifies the directory which should be set
* as the current working directory for the command.
@@ -1116,7 +1139,7 @@ public abstract class Shell {
* @param inheritParentEnv Indicates if the process should inherit the env
* vars from the parent process or not.
*/
- public ShellCommandExecutor(String[] execString, File dir,
+ public ShellCommandExecutor(String[] execString, File dir,
Map<String, String> env, long timeout, boolean inheritParentEnv) {
command = execString.clone();
if (dir != null) {
@@ -1141,7 +1164,7 @@ public abstract class Shell {
+ StringUtils.join(" ", command));
}
}
- this.run();
+ this.run();
}
@Override
@@ -1194,7 +1217,7 @@ public abstract class Shell {
/**
* To check if the passed script to shell command executor timed out or
* not.
- *
+ *
* @return if the script timed out.
*/
public boolean isTimedOut() {
@@ -1203,15 +1226,15 @@ public abstract class Shell {
/**
* Declare that the command has timed out.
- *
+ *
*/
private void setTimedOut() {
this.timedOut.set(true);
}
- /**
- * Static method to execute a shell command.
- * Covers most of the simple cases without requiring the user to implement
+ /**
+ * Static method to execute a shell command.
+ * Covers most of the simple cases without requiring the user to implement
* the <code>Shell</code> interface.
* @param cmd shell command to execute.
* @return the output of the executed command.
@@ -1233,7 +1256,7 @@ public abstract class Shell {
public static String execCommand(Map<String, String> env, String[] cmd,
long timeout) throws IOException {
- ShellCommandExecutor exec = new ShellCommandExecutor(cmd, null, env,
+ ShellCommandExecutor exec = new ShellCommandExecutor(cmd, null, env,
timeout);
exec.execute();
return exec.getOutput();
@@ -1271,7 +1294,7 @@ public abstract class Shell {
p.exitValue();
} catch (Exception e) {
//Process has not terminated.
- //So check if it has completed
+ //So check if it has completed
//if not just destroy it.
if (p != null && !shell.completed.get()) {
shell.setTimedOut();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/954465e7/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestShell.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestShell.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestShell.java
index 16ec867..acc7f1d 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestShell.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestShell.java
@@ -235,9 +235,9 @@ public class TestShell extends Assert {
expectedCommand =
new String[]{getWinUtilsPath(), "task", "isAlive", anyPid };
} else if (Shell.isSetsidAvailable) {
- expectedCommand = new String[] { "bash", "-c", "kill -0 -- -" + anyPid };
+ expectedCommand = new String[] {"kill", "-0", "--", "-" + anyPid };
} else {
- expectedCommand = new String[]{ "bash", "-c", "kill -0 " + anyPid };
+ expectedCommand = new String[] {"kill", "-0", anyPid };
}
Assert.assertArrayEquals(expectedCommand, checkProcessAliveCommand);
}
@@ -255,9 +255,9 @@ public class TestShell extends Assert {
expectedCommand =
new String[]{getWinUtilsPath(), "task", "kill", anyPid };
} else if (Shell.isSetsidAvailable) {
- expectedCommand = new String[] { "bash", "-c", "kill -9 -- -" + anyPid };
+ expectedCommand = new String[] {"kill", "-9", "--", "-" + anyPid };
} else {
- expectedCommand = new String[]{ "bash", "-c", "kill -9 " + anyPid };
+ expectedCommand = new String[] {"kill", "-9", anyPid };
}
Assert.assertArrayEquals(expectedCommand, checkProcessAliveCommand);
}
@@ -461,4 +461,10 @@ public class TestShell extends Assert {
}
}
+ @Test
+ public void testBashQuote() {
+ assertEquals("'foobar'", Shell.bashQuote("foobar"));
+ assertEquals("'foo'\\''bar'", Shell.bashQuote("foo'bar"));
+ assertEquals("''\\''foo'\\''bar'\\'''", Shell.bashQuote("'foo'bar'"));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[04/30] hadoop git commit: YARN-5443. Add support for docker inspect
command. Contributed by Shane Kumpf.
Posted by dr...@apache.org.
YARN-5443. Add support for docker inspect command. Contributed by Shane Kumpf.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/2e7c2a13
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/2e7c2a13
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/2e7c2a13
Branch: refs/heads/HADOOP-12756
Commit: 2e7c2a13a853b8195bc4f51f6c3c1f61656c2b33
Parents: a5fb298
Author: Varun Vasudev <vv...@apache.org>
Authored: Tue Aug 2 16:55:10 2016 +0530
Committer: Varun Vasudev <vv...@apache.org>
Committed: Tue Aug 2 16:55:10 2016 +0530
----------------------------------------------------------------------
.../runtime/docker/DockerInspectCommand.java | 41 ++++++++++++++++
.../docker/TestDockerInspectCommand.java | 50 ++++++++++++++++++++
2 files changed, 91 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2e7c2a13/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerInspectCommand.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerInspectCommand.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerInspectCommand.java
new file mode 100644
index 0000000..94b87b7
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerInspectCommand.java
@@ -0,0 +1,41 @@
+/*
+ * *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * /
+ */
+
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker;
+
+/**
+ * Encapsulates the docker inspect command and its command
+ * line arguments.
+ */
+public class DockerInspectCommand extends DockerCommand {
+ private static final String INSPECT_COMMAND = "inspect";
+ private String containerName;
+
+ public DockerInspectCommand(String containerName) {
+ super(INSPECT_COMMAND);
+ this.containerName = containerName;
+ }
+
+ public DockerInspectCommand getContainerStatus() {
+ super.addCommandArguments("--format='{{.State.Status}}'");
+ super.addCommandArguments(containerName);
+ return this;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2e7c2a13/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerInspectCommand.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerInspectCommand.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerInspectCommand.java
new file mode 100644
index 0000000..3321364
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerInspectCommand.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests the docker inspect command and its command
+ * line arguments.
+ */
+public class TestDockerInspectCommand {
+
+ private DockerInspectCommand dockerInspectCommand;
+
+ private static final String CONTAINER_NAME = "foo";
+
+ @Before
+ public void setup() {
+ dockerInspectCommand = new DockerInspectCommand(CONTAINER_NAME);
+ }
+
+ @Test
+ public void testGetCommandOption() {
+ assertEquals("inspect", dockerInspectCommand.getCommandOption());
+ }
+
+ @Test
+ public void testGetContainerStatus() throws Exception {
+ dockerInspectCommand.getContainerStatus();
+ assertEquals("inspect --format='{{.State.Status}}' foo",
+ dockerInspectCommand.getCommandWithArguments());
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[06/30] hadoop git commit: HADOOP-13429. Dispose of unnecessary SASL
servers. Contributed by Daryn Sharp.
Posted by dr...@apache.org.
HADOOP-13429. Dispose of unnecessary SASL servers. Contributed by Daryn Sharp.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/b3018e73
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/b3018e73
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/b3018e73
Branch: refs/heads/HADOOP-12756
Commit: b3018e73ccae43484d9cb85eabae814eb7f050a6
Parents: 7fc70c6
Author: Kihwal Lee <ki...@apache.org>
Authored: Tue Aug 2 10:40:28 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Tue Aug 2 10:40:28 2016 -0500
----------------------------------------------------------------------
.../src/main/java/org/apache/hadoop/ipc/Server.java | 13 +++++++++----
.../test/java/org/apache/hadoop/ipc/TestSaslRPC.java | 12 +++++++++++-
2 files changed, 20 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3018e73/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
index 405549a..80eea84 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
@@ -1598,7 +1598,10 @@ public abstract class Server {
String qop = (String) saslServer.getNegotiatedProperty(Sasl.QOP);
// SASL wrapping is only used if the connection has a QOP, and
// the value is not auth. ex. auth-int & auth-priv
- useWrap = (qop != null && !"auth".equalsIgnoreCase(qop));
+ useWrap = (qop != null && !"auth".equalsIgnoreCase(qop));
+ if (!useWrap) {
+ disposeSasl();
+ }
}
}
@@ -1692,9 +1695,9 @@ public abstract class Server {
private void switchToSimple() {
// disable SASL and blank out any SASL server
authProtocol = AuthProtocol.NONE;
- saslServer = null;
+ disposeSasl();
}
-
+
private RpcSaslProto buildSaslResponse(SaslState state, byte[] replyToken) {
if (LOG.isDebugEnabled()) {
LOG.debug("Will send " + state + " token of size "
@@ -1731,6 +1734,8 @@ public abstract class Server {
try {
saslServer.dispose();
} catch (SaslException ignored) {
+ } finally {
+ saslServer = null;
}
}
}
@@ -1980,7 +1985,7 @@ public abstract class Server {
.getProtocol() : null;
UserGroupInformation protocolUser = ProtoUtil.getUgi(connectionContext);
- if (saslServer == null) {
+ if (authProtocol == AuthProtocol.NONE) {
user = protocolUser;
} else {
// user is authenticated
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b3018e73/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestSaslRPC.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestSaslRPC.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestSaslRPC.java
index ec53e8c..72371a7 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestSaslRPC.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestSaslRPC.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.Client.ConnectionId;
+import org.apache.hadoop.ipc.Server.Connection;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.*;
import org.apache.hadoop.security.SaslRpcServer.AuthMethod;
@@ -270,7 +271,16 @@ public class TestSaslRPC extends TestRpcBase {
assertEquals(TOKEN, authMethod);
//QOP must be auth
assertEquals(expectedQop.saslQop,
- RPC.getConnectionIdForProxy(proxy).getSaslQop());
+ RPC.getConnectionIdForProxy(proxy).getSaslQop());
+ int n = 0;
+ for (Connection connection : server.getConnections()) {
+ // only qop auth should dispose of the sasl server
+ boolean hasServer = (connection.saslServer != null);
+ assertTrue("qop:" + expectedQop + " hasServer:" + hasServer,
+ (expectedQop == QualityOfProtection.AUTHENTICATION) ^ hasServer);
+ n++;
+ }
+ assertTrue(n > 0);
proxy.ping(null, newEmptyRequest());
} finally {
stop(server, proxy);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[15/30] hadoop git commit: HADOOP-13426. More efficiently build IPC
responses. Contributed by Daryn Sharp.
Posted by dr...@apache.org.
HADOOP-13426. More efficiently build IPC responses. Contributed by Daryn Sharp.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/2d822760
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/2d822760
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/2d822760
Branch: refs/heads/HADOOP-12756
Commit: 2d8227605fe22c1c05f31729edc8939013763c05
Parents: d848184
Author: Kihwal Lee <ki...@apache.org>
Authored: Wed Aug 3 09:30:24 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Wed Aug 3 09:31:49 2016 -0500
----------------------------------------------------------------------
.../org/apache/hadoop/ipc/ResponseBuffer.java | 98 +++++++++++++
.../main/java/org/apache/hadoop/ipc/Server.java | 143 +++++++++----------
.../apache/hadoop/ipc/TestResponseBuffer.java | 87 +++++++++++
3 files changed, 250 insertions(+), 78 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2d822760/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ResponseBuffer.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ResponseBuffer.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ResponseBuffer.java
new file mode 100644
index 0000000..ac96a24
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ResponseBuffer.java
@@ -0,0 +1,98 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ipc;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Arrays;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+@InterfaceAudience.Private
+class ResponseBuffer extends DataOutputStream {
+ ResponseBuffer(int capacity) {
+ super(new FramedBuffer(capacity));
+ }
+
+ // update framing bytes based on bytes written to stream.
+ private FramedBuffer getFramedBuffer() {
+ FramedBuffer buf = (FramedBuffer)out;
+ buf.setSize(written);
+ return buf;
+ }
+
+ void writeTo(OutputStream out) throws IOException {
+ getFramedBuffer().writeTo(out);
+ }
+
+ byte[] toByteArray() {
+ return getFramedBuffer().toByteArray();
+ }
+
+ int capacity() {
+ return ((FramedBuffer)out).capacity();
+ }
+
+ void setCapacity(int capacity) {
+ ((FramedBuffer)out).setCapacity(capacity);
+ }
+
+ void ensureCapacity(int capacity) {
+ if (((FramedBuffer)out).capacity() < capacity) {
+ ((FramedBuffer)out).setCapacity(capacity);
+ }
+ }
+
+ ResponseBuffer reset() {
+ written = 0;
+ ((FramedBuffer)out).reset();
+ return this;
+ }
+
+ private static class FramedBuffer extends ByteArrayOutputStream {
+ private static final int FRAMING_BYTES = 4;
+ FramedBuffer(int capacity) {
+ super(capacity + FRAMING_BYTES);
+ reset();
+ }
+ @Override
+ public int size() {
+ return count - FRAMING_BYTES;
+ }
+ void setSize(int size) {
+ buf[0] = (byte)((size >>> 24) & 0xFF);
+ buf[1] = (byte)((size >>> 16) & 0xFF);
+ buf[2] = (byte)((size >>> 8) & 0xFF);
+ buf[3] = (byte)((size >>> 0) & 0xFF);
+ }
+ int capacity() {
+ return buf.length - FRAMING_BYTES;
+ }
+ void setCapacity(int capacity) {
+ buf = Arrays.copyOf(buf, capacity + FRAMING_BYTES);
+ }
+ @Override
+ public void reset() {
+ count = FRAMING_BYTES;
+ setSize(0);
+ }
+ };
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2d822760/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
index 80eea84..4004fc0 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
@@ -80,12 +80,11 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configuration.IntegerRanges;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
-import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
-import org.apache.hadoop.ipc.ProtobufRpcEngine.RpcResponseMessageWrapper;
import org.apache.hadoop.ipc.ProtobufRpcEngine.RpcResponseWrapper;
+import org.apache.hadoop.ipc.ProtobufRpcEngine.RpcWrapper;
import org.apache.hadoop.ipc.RPC.RpcInvoker;
import org.apache.hadoop.ipc.RPC.VersionMismatch;
import org.apache.hadoop.ipc.metrics.RpcDetailedMetrics;
@@ -423,6 +422,13 @@ public abstract class Server {
private int maxQueueSize;
private final int maxRespSize;
+ private final ThreadLocal<ResponseBuffer> responseBuffer =
+ new ThreadLocal<ResponseBuffer>(){
+ @Override
+ protected ResponseBuffer initialValue() {
+ return new ResponseBuffer(INITIAL_RESP_BUF_SIZE);
+ }
+ };
private int socketSendBufferSize;
private final int maxDataLength;
private final boolean tcpNoDelay; // if T then disable Nagle's Algorithm
@@ -744,14 +750,7 @@ public abstract class Server {
public void abortResponse(Throwable t) throws IOException {
// don't send response if the call was already sent or aborted.
if (responseWaitCount.getAndSet(-1) > 0) {
- // clone the call to prevent a race with the other thread stomping
- // on the response while being sent. the original call is
- // effectively discarded since the wait count won't hit zero
- Call call = new Call(this);
- setupResponse(new ByteArrayOutputStream(), call,
- RpcStatusProto.FATAL, RpcErrorCodeProto.ERROR_RPC_SERVER,
- null, t.getClass().getName(), StringUtils.stringifyException(t));
- call.sendResponse();
+ connection.abortResponse(this, t);
}
}
@@ -1272,9 +1271,7 @@ public abstract class Server {
// must only wrap before adding to the responseQueue to prevent
// postponed responses from being encrypted and sent out of order.
if (call.connection.useWrap) {
- ByteArrayOutputStream response = new ByteArrayOutputStream();
- wrapWithSasl(response, call);
- call.setResponse(ByteBuffer.wrap(response.toByteArray()));
+ wrapWithSasl(call);
}
call.connection.responseQueue.addLast(call);
if (call.connection.responseQueue.size() == 1) {
@@ -1395,8 +1392,7 @@ public abstract class Server {
// Fake 'call' for failed authorization response
private final Call authFailedCall = new Call(AUTHORIZATION_FAILED_CALL_ID,
RpcConstants.INVALID_RETRY_COUNT, null, this);
- private ByteArrayOutputStream authFailedResponse = new ByteArrayOutputStream();
-
+
private boolean sentNegotiate = false;
private boolean useWrap = false;
@@ -1715,15 +1711,14 @@ public abstract class Server {
private void doSaslReply(Message message) throws IOException {
final Call saslCall = new Call(AuthProtocol.SASL.callId,
RpcConstants.INVALID_RETRY_COUNT, null, this);
- final ByteArrayOutputStream saslResponse = new ByteArrayOutputStream();
- setupResponse(saslResponse, saslCall,
+ setupResponse(saslCall,
RpcStatusProto.SUCCESS, null,
new RpcResponseWrapper(message), null, null);
saslCall.sendResponse();
}
private void doSaslReply(Exception ioe) throws IOException {
- setupResponse(authFailedResponse, authFailedCall,
+ setupResponse(authFailedCall,
RpcStatusProto.FATAL, RpcErrorCodeProto.FATAL_UNAUTHORIZED,
null, ioe.getClass().getName(), ioe.getLocalizedMessage());
authFailedCall.sendResponse();
@@ -1934,7 +1929,7 @@ public abstract class Server {
// Versions >>9 understand the normal response
Call fakeCall = new Call(-1, RpcConstants.INVALID_RETRY_COUNT, null,
this);
- setupResponse(buffer, fakeCall,
+ setupResponse(fakeCall,
RpcStatusProto.FATAL, RpcErrorCodeProto.FATAL_VERSION_MISMATCH,
null, VersionMismatch.class.getName(), errMsg);
fakeCall.sendResponse();
@@ -2111,7 +2106,7 @@ public abstract class Server {
} catch (WrappedRpcServerException wrse) { // inform client of error
Throwable ioe = wrse.getCause();
final Call call = new Call(callId, retry, null, this);
- setupResponse(authFailedResponse, call,
+ setupResponse(call,
RpcStatusProto.FATAL, wrse.getRpcErrorCodeProto(), null,
ioe.getClass().getName(), ioe.getMessage());
call.sendResponse();
@@ -2341,6 +2336,17 @@ public abstract class Server {
responder.doRespond(call);
}
+ private void abortResponse(Call call, Throwable t) throws IOException {
+ // clone the call to prevent a race with the other thread stomping
+ // on the response while being sent. the original call is
+ // effectively discarded since the wait count won't hit zero
+ call = new Call(call);
+ setupResponse(call,
+ RpcStatusProto.FATAL, RpcErrorCodeProto.ERROR_RPC_SERVER,
+ null, t.getClass().getName(), StringUtils.stringifyException(t));
+ call.sendResponse();
+ }
+
/**
* Get service class for connection
* @return the serviceClass
@@ -2384,8 +2390,6 @@ public abstract class Server {
public void run() {
LOG.debug(Thread.currentThread().getName() + ": starting");
SERVER.set(Server.this);
- ByteArrayOutputStream buf =
- new ByteArrayOutputStream(INITIAL_RESP_BUF_SIZE);
while (running) {
TraceScope traceScope = null;
try {
@@ -2455,16 +2459,8 @@ public abstract class Server {
}
CurCall.set(null);
synchronized (call.connection.responseQueue) {
- setupResponse(buf, call, returnStatus, detailedErr,
+ setupResponse(call, returnStatus, detailedErr,
value, errorClass, error);
-
- // Discard the large buf and reset it back to smaller size
- // to free up heap.
- if (buf.size() > maxRespSize) {
- LOG.warn("Large response size " + buf.size() + " for call "
- + call.toString());
- buf = new ByteArrayOutputStream(INITIAL_RESP_BUF_SIZE);
- }
call.sendResponse();
}
} catch (InterruptedException e) {
@@ -2682,13 +2678,11 @@ public abstract class Server {
* @param error error message, if the call failed
* @throws IOException
*/
- private static void setupResponse(ByteArrayOutputStream responseBuf,
- Call call, RpcStatusProto status, RpcErrorCodeProto erCode,
- Writable rv, String errorClass, String error)
- throws IOException {
- responseBuf.reset();
- DataOutputStream out = new DataOutputStream(responseBuf);
- RpcResponseHeaderProto.Builder headerBuilder =
+ private void setupResponse(
+ Call call, RpcStatusProto status, RpcErrorCodeProto erCode,
+ Writable rv, String errorClass, String error)
+ throws IOException {
+ RpcResponseHeaderProto.Builder headerBuilder =
RpcResponseHeaderProto.newBuilder();
headerBuilder.setClientId(ByteString.copyFrom(call.clientId));
headerBuilder.setCallId(call.callId);
@@ -2698,32 +2692,14 @@ public abstract class Server {
if (status == RpcStatusProto.SUCCESS) {
RpcResponseHeaderProto header = headerBuilder.build();
- final int headerLen = header.getSerializedSize();
- int fullLength = CodedOutputStream.computeRawVarint32Size(headerLen) +
- headerLen;
try {
- if (rv instanceof ProtobufRpcEngine.RpcWrapper) {
- ProtobufRpcEngine.RpcWrapper resWrapper =
- (ProtobufRpcEngine.RpcWrapper) rv;
- fullLength += resWrapper.getLength();
- out.writeInt(fullLength);
- header.writeDelimitedTo(out);
- rv.write(out);
- } else { // Have to serialize to buffer to get len
- final DataOutputBuffer buf = new DataOutputBuffer();
- rv.write(buf);
- byte[] data = buf.getData();
- fullLength += buf.getLength();
- out.writeInt(fullLength);
- header.writeDelimitedTo(out);
- out.write(data, 0, buf.getLength());
- }
+ setupResponse(call, header, rv);
} catch (Throwable t) {
LOG.warn("Error serializing call response for call " + call, t);
// Call back to same function - this is OK since the
// buffer is reset at the top, and since status is changed
// to ERROR it won't infinite loop.
- setupResponse(responseBuf, call, RpcStatusProto.ERROR,
+ setupResponse(call, RpcStatusProto.ERROR,
RpcErrorCodeProto.ERROR_SERIALIZING_RESPONSE,
null, t.getClass().getName(),
StringUtils.stringifyException(t));
@@ -2733,16 +2709,35 @@ public abstract class Server {
headerBuilder.setExceptionClassName(errorClass);
headerBuilder.setErrorMsg(error);
headerBuilder.setErrorDetail(erCode);
- RpcResponseHeaderProto header = headerBuilder.build();
- int headerLen = header.getSerializedSize();
- final int fullLength =
- CodedOutputStream.computeRawVarint32Size(headerLen) + headerLen;
- out.writeInt(fullLength);
- header.writeDelimitedTo(out);
+ setupResponse(call, headerBuilder.build(), null);
}
- call.setResponse(ByteBuffer.wrap(responseBuf.toByteArray()));
}
-
+
+ private void setupResponse(Call call,
+ RpcResponseHeaderProto header, Writable rv) throws IOException {
+ ResponseBuffer buf = responseBuffer.get().reset();
+ // adjust capacity on estimated length to reduce resizing copies
+ int estimatedLen = header.getSerializedSize();
+ estimatedLen += CodedOutputStream.computeRawVarint32Size(estimatedLen);
+ // if it's not a wrapped protobuf, just let it grow on its own
+ if (rv instanceof RpcWrapper) {
+ estimatedLen += ((RpcWrapper)rv).getLength();
+ }
+ buf.ensureCapacity(estimatedLen);
+ header.writeDelimitedTo(buf);
+ if (rv != null) { // null for exceptions
+ rv.write(buf);
+ }
+ call.setResponse(ByteBuffer.wrap(buf.toByteArray()));
+ // Discard a large buf and reset it back to smaller size
+ // to free up heap.
+ if (buf.capacity() > maxRespSize) {
+ LOG.warn("Large response size " + buf.size() + " for call "
+ + call.toString());
+ buf.setCapacity(INITIAL_RESP_BUF_SIZE);
+ }
+ }
+
/**
* Setup response for the IPC Call on Fatal Error from a
* client that is using old version of Hadoop.
@@ -2769,10 +2764,8 @@ public abstract class Server {
WritableUtils.writeString(out, error);
call.setResponse(ByteBuffer.wrap(response.toByteArray()));
}
-
-
- private static void wrapWithSasl(ByteArrayOutputStream response, Call call)
- throws IOException {
+
+ private void wrapWithSasl(Call call) throws IOException {
if (call.connection.saslServer != null) {
byte[] token = call.rpcResponse.array();
// synchronization may be needed since there can be multiple Handler
@@ -2783,7 +2776,6 @@ public abstract class Server {
if (LOG.isDebugEnabled())
LOG.debug("Adding saslServer wrapped token of size " + token.length
+ " as call response.");
- response.reset();
// rebuild with sasl header and payload
RpcResponseHeaderProto saslHeader = RpcResponseHeaderProto.newBuilder()
.setCallId(AuthProtocol.SASL.callId)
@@ -2791,14 +2783,9 @@ public abstract class Server {
.build();
RpcSaslProto saslMessage = RpcSaslProto.newBuilder()
.setState(SaslState.WRAP)
- .setToken(ByteString.copyFrom(token, 0, token.length))
+ .setToken(ByteString.copyFrom(token))
.build();
- RpcResponseMessageWrapper saslResponse =
- new RpcResponseMessageWrapper(saslHeader, saslMessage);
-
- DataOutputStream out = new DataOutputStream(response);
- out.writeInt(saslResponse.getLength());
- saslResponse.write(out);
+ setupResponse(call, saslHeader, new RpcResponseWrapper(saslMessage));
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2d822760/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestResponseBuffer.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestResponseBuffer.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestResponseBuffer.java
new file mode 100644
index 0000000..98743be
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestResponseBuffer.java
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ipc;
+
+import static org.junit.Assert.assertEquals;
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import org.apache.hadoop.ipc.ResponseBuffer;
+import org.junit.Test;
+
+/** Unit tests for ResponseBuffer. */
+public class TestResponseBuffer {
+ @Test
+ public void testBuffer() throws IOException {
+ final int startSize = 8;
+ final String empty = "";
+ ResponseBuffer buf = new ResponseBuffer(startSize);
+ assertEquals(startSize, buf.capacity());
+
+ // verify it's initially empty
+ checkBuffer(buf, empty);
+ // write "nothing" and re-verify it's empty
+ buf.writeBytes(empty);
+ checkBuffer(buf, empty);
+
+ // write to the buffer twice and verify it's properly encoded
+ String s1 = "testing123";
+ buf.writeBytes(s1);
+ checkBuffer(buf, s1);
+ String s2 = "456!";
+ buf.writeBytes(s2);
+ checkBuffer(buf, s1 + s2);
+
+ // reset should not change length of underlying byte array
+ int length = buf.capacity();
+ buf.reset();
+ assertEquals(length, buf.capacity());
+ checkBuffer(buf, empty);
+
+ // setCapacity will change length of underlying byte array
+ buf.setCapacity(startSize);
+ assertEquals(startSize, buf.capacity());
+ checkBuffer(buf, empty);
+
+ // make sure it still works
+ buf.writeBytes(s1);
+ checkBuffer(buf, s1);
+ buf.writeBytes(s2);
+ checkBuffer(buf, s1 + s2);
+ }
+
+ private void checkBuffer(ResponseBuffer buf, String expected)
+ throws IOException {
+ // buffer payload length matches expected length
+ int expectedLength = expected.getBytes().length;
+ assertEquals(expectedLength, buf.size());
+ // buffer has the framing bytes (int)
+ byte[] framed = buf.toByteArray();
+ assertEquals(expectedLength + 4, framed.length);
+
+ // verify encoding of buffer: framing (int) + payload bytes
+ DataInputStream dis =
+ new DataInputStream(new ByteArrayInputStream(framed));
+ assertEquals(expectedLength, dis.readInt());
+ assertEquals(expectedLength, dis.available());
+ byte[] payload = new byte[expectedLength];
+ dis.readFully(payload);
+ assertEquals(expected, new String(payload));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[08/30] hadoop git commit: HADOOP-13081. add the ability to create
multiple UGIs/subjects from one kerberos login. Contributed by Sergey
Shelukhin.
Posted by dr...@apache.org.
HADOOP-13081. add the ability to create multiple UGIs/subjects from one kerberos login. Contributed by Sergey Shelukhin.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/0458a2af
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/0458a2af
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/0458a2af
Branch: refs/heads/HADOOP-12756
Commit: 0458a2af6e925d023882714e8b7b0568eca7a775
Parents: 3818393
Author: Chris Nauroth <cn...@apache.org>
Authored: Tue Aug 2 12:43:30 2016 -0700
Committer: Chris Nauroth <cn...@apache.org>
Committed: Tue Aug 2 12:43:30 2016 -0700
----------------------------------------------------------------------
.../hadoop/security/UserGroupInformation.java | 29 +++++++++++++++++++-
.../security/TestUserGroupInformation.java | 27 ++++++++++++++++++
2 files changed, 55 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0458a2af/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java
index 7dee14b..1e5ad36 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java
@@ -38,6 +38,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
@@ -637,7 +638,33 @@ public class UserGroupInformation {
this.isKeytab = KerberosUtil.hasKerberosKeyTab(subject);
this.isKrbTkt = KerberosUtil.hasKerberosTicket(subject);
}
-
+
+ /**
+ * Copies the Subject of this UGI and creates a new UGI with the new subject.
+ * This can be used to add credentials (e.g. tokens) to different copies of
+ * the same UGI, allowing multiple users with different tokens to reuse the
+ * UGI without re-authenticating with Kerberos.
+ * @return clone of the UGI with a new subject.
+ */
+ @InterfaceAudience.Public
+ @InterfaceStability.Evolving
+ public UserGroupInformation copySubjectAndUgi() {
+ Subject subj = getSubject();
+ // The ctor will set other fields automatically from the principals.
+ return new UserGroupInformation(new Subject(false, subj.getPrincipals(),
+ cloneCredentials(subj.getPublicCredentials()),
+ cloneCredentials(subj.getPrivateCredentials())));
+ }
+
+ private static Set<Object> cloneCredentials(Set<Object> old) {
+ Set<Object> set = new HashSet<>();
+ // Make sure Hadoop credentials objects do not reuse the maps.
+ for (Object o : old) {
+ set.add(o instanceof Credentials ? new Credentials((Credentials)o) : o);
+ }
+ return set;
+ }
+
/**
* checks if logged in using kerberos
* @return true if the subject logged via keytab or has a Kerberos TGT
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0458a2af/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/TestUserGroupInformation.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/TestUserGroupInformation.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/TestUserGroupInformation.java
index 91f36e5..8c41180 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/TestUserGroupInformation.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/TestUserGroupInformation.java
@@ -44,6 +44,7 @@ import java.security.PrivilegedExceptionAction;
import java.util.Collection;
import java.util.ConcurrentModificationException;
import java.util.LinkedHashSet;
+import java.util.List;
import java.util.Set;
import static org.apache.hadoop.fs.CommonConfigurationKeys.HADOOP_USER_GROUP_METRICS_PERCENTILES_INTERVALS;
@@ -877,6 +878,32 @@ public class TestUserGroupInformation {
assertEquals(1, tokens.size());
}
+ @Test(timeout = 30000)
+ public void testCopySubjectAndUgi() throws IOException {
+ SecurityUtil.setAuthenticationMethod(AuthenticationMethod.SIMPLE, conf);
+ UserGroupInformation.setConfiguration(conf);
+ UserGroupInformation u1 = UserGroupInformation.getLoginUser();
+ assertNotNull(u1);
+ @SuppressWarnings("unchecked")
+ Token<? extends TokenIdentifier> tmpToken = mock(Token.class);
+ u1.addToken(tmpToken);
+
+ UserGroupInformation u2 = u1.copySubjectAndUgi();
+ assertEquals(u1.getAuthenticationMethod(), u2.getAuthenticationMethod());
+ assertNotSame(u1.getSubject(), u2.getSubject());
+ Credentials c1 = u1.getCredentials(), c2 = u2.getCredentials();
+ List<Text> sc1 = c1.getAllSecretKeys(), sc2 = c2.getAllSecretKeys();
+ assertArrayEquals(sc1.toArray(new Text[0]), sc2.toArray(new Text[0]));
+ Collection<Token<? extends TokenIdentifier>> ts1 = c1.getAllTokens(),
+ ts2 = c2.getAllTokens();
+ assertArrayEquals(ts1.toArray(new Token[0]), ts2.toArray(new Token[0]));
+ @SuppressWarnings("unchecked")
+ Token<? extends TokenIdentifier> token = mock(Token.class);
+ u2.addToken(token);
+ assertTrue(u2.getCredentials().getAllTokens().contains(token));
+ assertFalse(u1.getCredentials().getAllTokens().contains(token));
+ }
+
/**
* This test checks a race condition between getting and adding tokens for
* the current user. Calling UserGroupInformation.getCurrentUser() returns
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[03/30] hadoop git commit: HDFS-10707. Replace
org.apache.commons.io.Charsets with java.nio.charset.StandardCharsets.
Contributed by Vincent Poon.
Posted by dr...@apache.org.
HDFS-10707. Replace org.apache.commons.io.Charsets with java.nio.charset.StandardCharsets. Contributed by Vincent Poon.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/a5fb298e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/a5fb298e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/a5fb298e
Branch: refs/heads/HADOOP-12756
Commit: a5fb298e56220a35d61b8d2bda716d8fb8ef8bb7
Parents: 6890d5b
Author: Akira Ajisaka <aa...@apache.org>
Authored: Tue Aug 2 17:07:59 2016 +0900
Committer: Akira Ajisaka <aa...@apache.org>
Committed: Tue Aug 2 17:07:59 2016 +0900
----------------------------------------------------------------------
.../src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java | 4 ++--
.../java/org/apache/hadoop/fs/http/client/HttpFSUtils.java | 5 +++--
.../hadoop/fs/http/server/HttpFSAuthenticationFilter.java | 4 ++--
.../java/org/apache/hadoop/lib/wsrs/JSONMapProvider.java | 5 +++--
.../main/java/org/apache/hadoop/lib/wsrs/JSONProvider.java | 5 +++--
.../hdfs/server/datanode/web/webhdfs/ParameterParser.java | 9 +++++----
.../hdfs/server/datanode/web/webhdfs/WebHdfsHandler.java | 5 +++--
.../apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java | 7 ++++---
.../hadoop/hdfs/server/namenode/FSDirStatAndListingOp.java | 5 +++--
.../hadoop/hdfs/server/namenode/FSDirWriteFileOp.java | 4 ++--
.../org/apache/hadoop/hdfs/server/namenode/FSImageUtil.java | 4 ++--
.../hadoop/hdfs/server/namenode/TestGetBlockLocations.java | 4 ++--
12 files changed, 34 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java
index d7faa68..111c8c4 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.hdfs;
import com.google.common.base.Joiner;
import com.google.common.collect.Maps;
import com.google.common.primitives.SignedBytes;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.crypto.key.KeyProvider;
import org.apache.hadoop.fs.BlockLocation;
@@ -71,6 +70,7 @@ import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.URI;
import java.nio.channels.SocketChannel;
+import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.Collection;
import java.util.Collections;
@@ -95,7 +95,7 @@ public class DFSUtilClient {
* Converts a string to a byte array using UTF8 encoding.
*/
public static byte[] string2Bytes(String str) {
- return str.getBytes(Charsets.UTF_8);
+ return str.getBytes(StandardCharsets.UTF_8);
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/client/HttpFSUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/client/HttpFSUtils.java b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/client/HttpFSUtils.java
index 45d9805..95e26d7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/client/HttpFSUtils.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/client/HttpFSUtils.java
@@ -17,7 +17,6 @@
*/
package org.apache.hadoop.fs.http.client;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.fs.Path;
import org.json.simple.parser.JSONParser;
@@ -29,6 +28,7 @@ import java.net.HttpURLConnection;
import java.net.URI;
import java.net.URL;
import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
import java.text.MessageFormat;
import java.util.List;
import java.util.Map;
@@ -128,7 +128,8 @@ public class HttpFSUtils {
static Object jsonParse(HttpURLConnection conn) throws IOException {
try {
JSONParser parser = new JSONParser();
- return parser.parse(new InputStreamReader(conn.getInputStream(), Charsets.UTF_8));
+ return parser.parse(
+ new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8));
} catch (ParseException ex) {
throw new IOException("JSON parser error, " + ex.getMessage(), ex);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSAuthenticationFilter.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSAuthenticationFilter.java b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSAuthenticationFilter.java
index a68ed3b..aa61ef7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSAuthenticationFilter.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSAuthenticationFilter.java
@@ -17,7 +17,6 @@
*/
package org.apache.hadoop.fs.http.server;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.web.WebHdfsConstants;
@@ -32,6 +31,7 @@ import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
+import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Properties;
@@ -83,7 +83,7 @@ public class HttpFSAuthenticationFilter
try {
StringBuilder secret = new StringBuilder();
Reader reader = new InputStreamReader(new FileInputStream(
- signatureSecretFile), Charsets.UTF_8);
+ signatureSecretFile), StandardCharsets.UTF_8);
int c = reader.read();
while (c > -1) {
secret.append((char)c);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONMapProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONMapProvider.java b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONMapProvider.java
index 0809a85..05bb9a1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONMapProvider.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONMapProvider.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.lib.wsrs;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.classification.InterfaceAudience;
import org.json.simple.JSONObject;
@@ -34,6 +33,7 @@ import java.io.OutputStreamWriter;
import java.io.Writer;
import java.lang.annotation.Annotation;
import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
import java.util.Map;
@Provider
@@ -56,7 +56,8 @@ public class JSONMapProvider implements MessageBodyWriter<Map> {
public void writeTo(Map map, Class<?> aClass, Type type, Annotation[] annotations,
MediaType mediaType, MultivaluedMap<String, Object> stringObjectMultivaluedMap,
OutputStream outputStream) throws IOException, WebApplicationException {
- Writer writer = new OutputStreamWriter(outputStream, Charsets.UTF_8);
+ Writer writer =
+ new OutputStreamWriter(outputStream, StandardCharsets.UTF_8);
JSONObject.writeJSONString(map, writer);
writer.write(ENTER);
writer.flush();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONProvider.java b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONProvider.java
index 15ea8e6..7e1f98b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONProvider.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/lib/wsrs/JSONProvider.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.lib.wsrs;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.classification.InterfaceAudience;
import org.json.simple.JSONStreamAware;
@@ -34,6 +33,7 @@ import java.io.OutputStreamWriter;
import java.io.Writer;
import java.lang.annotation.Annotation;
import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
@Provider
@Produces(MediaType.APPLICATION_JSON)
@@ -56,7 +56,8 @@ public class JSONProvider implements MessageBodyWriter<JSONStreamAware> {
public void writeTo(JSONStreamAware jsonStreamAware, Class<?> aClass, Type type, Annotation[] annotations,
MediaType mediaType, MultivaluedMap<String, Object> stringObjectMultivaluedMap,
OutputStream outputStream) throws IOException, WebApplicationException {
- Writer writer = new OutputStreamWriter(outputStream, Charsets.UTF_8);
+ Writer writer =
+ new OutputStreamWriter(outputStream, StandardCharsets.UTF_8);
jsonStreamAware.writeJSONString(writer);
writer.write(ENTER);
writer.flush();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/ParameterParser.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/ParameterParser.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/ParameterParser.java
index 440a532..3c7016e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/ParameterParser.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/ParameterParser.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.hdfs.server.datanode.web.webhdfs;
import io.netty.handler.codec.http.QueryStringDecoder;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.permission.FsPermission;
@@ -45,6 +44,7 @@ import org.apache.hadoop.security.token.Token;
import java.io.IOException;
import java.net.URI;
import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
@@ -58,8 +58,8 @@ class ParameterParser {
private final Map<String, List<String>> params;
ParameterParser(QueryStringDecoder decoder, Configuration conf) {
- this.path = decodeComponent(decoder.path().substring
- (WEBHDFS_PREFIX_LENGTH), Charsets.UTF_8);
+ this.path = decodeComponent(decoder.path().substring(WEBHDFS_PREFIX_LENGTH),
+ StandardCharsets.UTF_8);
this.params = decoder.parameters();
this.conf = conf;
}
@@ -137,7 +137,8 @@ class ParameterParser {
}
public EnumSet<CreateFlag> createFlag() {
- String cf = decodeComponent(param(CreateFlagParam.NAME), Charsets.UTF_8);
+ String cf =
+ decodeComponent(param(CreateFlagParam.NAME), StandardCharsets.UTF_8);
return new CreateFlagParam(cf).getValue();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/WebHdfsHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/WebHdfsHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/WebHdfsHandler.java
index 0a8f40d..d354612 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/WebHdfsHandler.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/webhdfs/WebHdfsHandler.java
@@ -57,10 +57,10 @@ import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
import java.security.PrivilegedExceptionAction;
import java.util.EnumSet;
-import org.apache.commons.io.Charsets;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -283,7 +283,8 @@ public class WebHdfsHandler extends SimpleChannelInboundHandler<HttpRequest> {
} finally {
IOUtils.cleanup(LOG, dfsclient);
}
- final byte[] js = JsonUtil.toJsonString(checksum).getBytes(Charsets.UTF_8);
+ final byte[] js =
+ JsonUtil.toJsonString(checksum).getBytes(StandardCharsets.UTF_8);
resp =
new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(js));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java
index f51427f..ae73f9c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.hdfs.server.namenode;
import com.google.common.base.Preconditions;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.InvalidPathException;
import org.apache.hadoop.fs.UnresolvedLinkException;
@@ -33,6 +32,7 @@ import org.apache.hadoop.hdfs.protocol.QuotaExceededException;
import org.apache.hadoop.hdfs.server.namenode.snapshot.Snapshot;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.AbstractMap;
import java.util.List;
import java.util.Map;
@@ -121,7 +121,8 @@ class FSDirMkdirOp {
static Map.Entry<INodesInPath, String> createAncestorDirectories(
FSDirectory fsd, INodesInPath iip, PermissionStatus permission)
throws IOException {
- final String last = new String(iip.getLastLocalName(), Charsets.UTF_8);
+ final String last =
+ new String(iip.getLastLocalName(), StandardCharsets.UTF_8);
INodesInPath existing = iip.getExistingINodes();
List<String> children = iip.getPath(existing.length(),
iip.length() - existing.length());
@@ -189,7 +190,7 @@ class FSDirMkdirOp {
throws IOException {
assert fsd.hasWriteLock();
existing = unprotectedMkdir(fsd, fsd.allocateNewInodeId(), existing,
- localName.getBytes(Charsets.UTF_8), perm, null, now());
+ localName.getBytes(StandardCharsets.UTF_8), perm, null, now());
if (existing == null) {
return null;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirStatAndListingOp.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirStatAndListingOp.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirStatAndListingOp.java
index 3acfcc9..0604b9a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirStatAndListingOp.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirStatAndListingOp.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.hdfs.server.namenode;
import com.google.common.base.Preconditions;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.fs.ContentSummary;
import org.apache.hadoop.fs.DirectoryListingStartAfterNotFoundException;
import org.apache.hadoop.fs.FileEncryptionInfo;
@@ -46,6 +45,7 @@ import org.apache.hadoop.hdfs.util.ReadOnlyList;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import static org.apache.hadoop.util.Time.now;
@@ -55,7 +55,8 @@ class FSDirStatAndListingOp {
byte[] startAfter, boolean needLocation) throws IOException {
byte[][] pathComponents = FSDirectory
.getPathComponentsForReservedPath(srcArg);
- final String startAfterString = new String(startAfter, Charsets.UTF_8);
+ final String startAfterString =
+ new String(startAfter, StandardCharsets.UTF_8);
String src = null;
if (fsd.isPermissionEnabled()) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java
index 01743ba..d428c95 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.hdfs.server.namenode;
import com.google.common.base.Preconditions;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.crypto.CipherSuite;
import org.apache.hadoop.crypto.CryptoProtocolVersion;
@@ -60,6 +59,7 @@ import org.apache.hadoop.util.ChunkedArrayList;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -610,7 +610,7 @@ class FSDirWriteFileOp {
}
INodeFile newNode = newINodeFile(fsd.allocateNewInodeId(), permissions,
modTime, modTime, replication, preferredBlockSize, ecPolicy != null);
- newNode.setLocalName(localName.getBytes(Charsets.UTF_8));
+ newNode.setLocalName(localName.getBytes(StandardCharsets.UTF_8));
newNode.toUnderConstruction(clientName, clientMachine);
newiip = fsd.addINode(existing, newNode);
} finally {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageUtil.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageUtil.java
index 388a1bf..5dbc80c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageUtil.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageUtil.java
@@ -21,9 +21,9 @@ import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.protocol.LayoutVersion.Feature;
@@ -34,7 +34,7 @@ import org.apache.hadoop.io.compress.CompressionCodec;
@InterfaceAudience.Private
public final class FSImageUtil {
public static final byte[] MAGIC_HEADER =
- "HDFSIMG1".getBytes(Charsets.UTF_8);
+ "HDFSIMG1".getBytes(StandardCharsets.UTF_8);
public static final int FILE_VERSION = 1;
public static boolean checkFileFormat(RandomAccessFile file)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5fb298e/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestGetBlockLocations.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestGetBlockLocations.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestGetBlockLocations.java
index 7a3a8d6..eec5c98 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestGetBlockLocations.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestGetBlockLocations.java
@@ -17,7 +17,6 @@
*/
package org.apache.hadoop.hdfs.server.namenode;
-import org.apache.commons.io.Charsets;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.fs.permission.PermissionStatus;
@@ -27,6 +26,7 @@ import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_BLOCK_SIZE_DEFAULT;
@@ -124,7 +124,7 @@ public class TestGetBlockLocations {
"hdfs", "supergroup",
FsPermission.createImmutable((short) 0x1ff));
final INodeFile file = new INodeFile(
- MOCK_INODE_ID, FILE_NAME.getBytes(Charsets.UTF_8),
+ MOCK_INODE_ID, FILE_NAME.getBytes(StandardCharsets.UTF_8),
perm, 1, 1, new BlockInfo[] {}, (short) 1,
DFS_BLOCK_SIZE_DEFAULT);
fsn.getFSDirectory().addINode(iip, file);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[16/30] hadoop git commit: HDFS-10656. Optimize conversion of byte
arrays back to path string. Contributed by Daryn Sharp.
Posted by dr...@apache.org.
HDFS-10656. Optimize conversion of byte arrays back to path string. Contributed by Daryn Sharp.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/bebf10d2
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/bebf10d2
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/bebf10d2
Branch: refs/heads/HADOOP-12756
Commit: bebf10d2455cad1aa8985553417d4d74a61150ee
Parents: 2d82276
Author: Kihwal Lee <ki...@apache.org>
Authored: Wed Aug 3 11:53:41 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Wed Aug 3 11:53:41 2016 -0500
----------------------------------------------------------------------
.../java/org/apache/hadoop/hdfs/DFSUtil.java | 52 ++++++++++++--------
1 file changed, 32 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/bebf10d2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSUtil.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSUtil.java
index 5ab6978..3fe944e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSUtil.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSUtil.java
@@ -84,7 +84,6 @@ import org.apache.hadoop.security.authorize.AccessControlList;
import org.apache.hadoop.util.ToolRunner;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Charsets;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
@@ -267,27 +266,40 @@ public class DFSUtil {
/**
* Given a list of path components returns a path as a UTF8 String
*/
- public static String byteArray2PathString(byte[][] pathComponents,
- int offset, int length) {
- if (pathComponents.length == 0) {
+ public static String byteArray2PathString(final byte[][] components,
+ final int offset, final int length) {
+ // specifically not using StringBuilder to more efficiently build
+ // string w/o excessive byte[] copies and charset conversions.
+ final int range = offset + length;
+ Preconditions.checkPositionIndexes(offset, range, components.length);
+ if (length == 0) {
return "";
}
- Preconditions.checkArgument(offset >= 0 && offset < pathComponents.length);
- Preconditions.checkArgument(length >= 0 && offset + length <=
- pathComponents.length);
- if (offset == 0 && length == 1
- && (pathComponents[0] == null || pathComponents[0].length == 0)) {
- return Path.SEPARATOR;
- }
- StringBuilder result = new StringBuilder();
- int lastIndex = offset + length - 1;
- for (int i = offset; i <= lastIndex; i++) {
- result.append(new String(pathComponents[i], Charsets.UTF_8));
- if (i < lastIndex) {
- result.append(Path.SEPARATOR_CHAR);
- }
- }
- return result.toString();
+ // absolute paths start with either null or empty byte[]
+ byte[] firstComponent = components[offset];
+ boolean isAbsolute = (offset == 0 &&
+ (firstComponent == null || firstComponent.length == 0));
+ if (offset == 0 && length == 1) {
+ return isAbsolute ? Path.SEPARATOR : bytes2String(firstComponent);
+ }
+ // compute length of full byte[], seed with 1st component and delimiters
+ int pos = isAbsolute ? 0 : firstComponent.length;
+ int size = pos + length - 1;
+ for (int i=offset + 1; i < range; i++) {
+ size += components[i].length;
+ }
+ final byte[] result = new byte[size];
+ if (!isAbsolute) {
+ System.arraycopy(firstComponent, 0, result, 0, firstComponent.length);
+ }
+ // append remaining components as "/component".
+ for (int i=offset + 1; i < range; i++) {
+ result[pos++] = (byte)Path.SEPARATOR_CHAR;
+ int len = components[i].length;
+ System.arraycopy(components[i], 0, result, pos, len);
+ pos += len;
+ }
+ return bytes2String(result);
}
public static String byteArray2PathString(byte[][] pathComponents) {
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[09/30] hadoop git commit: HDFS-5805. TestCheckpoint.testCheckpoint
fails intermittently on branch-2. Contributed by Eric Badger.
Posted by dr...@apache.org.
HDFS-5805. TestCheckpoint.testCheckpoint fails intermittently on branch-2. Contributed by Eric Badger.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/5e5b8793
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/5e5b8793
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/5e5b8793
Branch: refs/heads/HADOOP-12756
Commit: 5e5b8793fba8e25aeba7a74878da4cf8e806f061
Parents: 0458a2a
Author: Kihwal Lee <ki...@apache.org>
Authored: Tue Aug 2 15:37:36 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Tue Aug 2 15:37:36 2016 -0500
----------------------------------------------------------------------
.../apache/hadoop/hdfs/server/namenode/TestCheckpoint.java | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5e5b8793/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestCheckpoint.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestCheckpoint.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestCheckpoint.java
index 7c9df29..e643d21 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestCheckpoint.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestCheckpoint.java
@@ -1030,6 +1030,7 @@ public class TestCheckpoint {
*/
@Test
public void testCheckpoint() throws IOException {
+ Path tmpDir = new Path("/tmp_tmp");
Path file1 = new Path("checkpoint.dat");
Path file2 = new Path("checkpoint2.dat");
Configuration conf = new HdfsConfiguration();
@@ -1057,6 +1058,11 @@ public class TestCheckpoint {
replication, seed);
checkFile(fileSys, file1, replication);
+ for(int i=0; i < 1000; i++) {
+ fileSys.mkdirs(tmpDir);
+ fileSys.delete(tmpDir, true);
+ }
+
//
// Take a checkpoint
//
@@ -1081,7 +1087,6 @@ public class TestCheckpoint {
//
// Restart cluster and verify that file1 still exist.
//
- Path tmpDir = new Path("/tmp_tmp");
try {
cluster = new MiniDFSCluster.Builder(conf).numDataNodes(numDatanodes)
.format(false).build();
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[23/30] hadoop git commit: YARN-5469. Increase timeout of
TestAmFilter.testFilter. Contributed by Eric Badger
Posted by dr...@apache.org.
YARN-5469. Increase timeout of TestAmFilter.testFilter. Contributed by Eric Badger
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/db4a61dc
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/db4a61dc
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/db4a61dc
Branch: refs/heads/HADOOP-12756
Commit: db4a61dc617954cf448a0d44cb4ac79f2abf9db3
Parents: f4ba5ff
Author: Jason Lowe <jl...@apache.org>
Authored: Wed Aug 3 19:50:31 2016 +0000
Committer: Jason Lowe <jl...@apache.org>
Committed: Wed Aug 3 19:51:44 2016 +0000
----------------------------------------------------------------------
.../apache/hadoop/yarn/server/webproxy/amfilter/TestAmFilter.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/db4a61dc/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/test/java/org/apache/hadoop/yarn/server/webproxy/amfilter/TestAmFilter.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/test/java/org/apache/hadoop/yarn/server/webproxy/amfilter/TestAmFilter.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/test/java/org/apache/hadoop/yarn/server/webproxy/amfilter/TestAmFilter.java
index 4601c20..6f64777 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/test/java/org/apache/hadoop/yarn/server/webproxy/amfilter/TestAmFilter.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/test/java/org/apache/hadoop/yarn/server/webproxy/amfilter/TestAmFilter.java
@@ -123,7 +123,7 @@ public class TestAmFilter {
/**
* Test AmIpFilter
*/
- @Test(timeout = 1000)
+ @Test(timeout = 10000)
@SuppressWarnings("deprecation")
public void testFilter() throws Exception {
Map<String, String> params = new HashMap<String, String>();
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[11/30] hadoop git commit: HDFS-9353. Code and comment mismatch in
JavaKeyStoreProvider. (Andras Bokor)
Posted by dr...@apache.org.
HDFS-9353. Code and comment mismatch in JavaKeyStoreProvider. (Andras Bokor)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/d28c2d9f
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/d28c2d9f
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/d28c2d9f
Branch: refs/heads/HADOOP-12756
Commit: d28c2d9f556372bce2d6a6d56fefabf0ba47094f
Parents: 954465e
Author: Arpit Agarwal <ar...@apache.org>
Authored: Tue Aug 2 15:17:45 2016 -0700
Committer: Arpit Agarwal <ar...@apache.org>
Committed: Tue Aug 2 15:17:45 2016 -0700
----------------------------------------------------------------------
.../src/main/java/org/apache/hadoop/security/ProviderUtils.java | 2 --
1 file changed, 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/d28c2d9f/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/ProviderUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/ProviderUtils.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/ProviderUtils.java
index 84ee912..013e56c 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/ProviderUtils.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/ProviderUtils.java
@@ -212,8 +212,6 @@ public final class ProviderUtils {
public static char[] locatePassword(String envWithPass, String fileWithPass)
throws IOException {
char[] pass = null;
- // Get the password file from the conf, if not present from the user's
- // environment var
if (System.getenv().containsKey(envWithPass)) {
pass = System.getenv(envWithPass).toCharArray();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[13/30] hadoop git commit: HADOOP-13459. hadoop-azure runs several
test cases repeatedly,
causing unnecessarily long running time. Contributed by Chris Nauroth.
Posted by dr...@apache.org.
HADOOP-13459. hadoop-azure runs several test cases repeatedly, causing unnecessarily long running time. Contributed by Chris Nauroth.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/4ef1bc0b
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/4ef1bc0b
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/4ef1bc0b
Branch: refs/heads/HADOOP-12756
Commit: 4ef1bc0bdb6b2c907431492e4c8395035fa96c84
Parents: b913677
Author: Chris Nauroth <cn...@apache.org>
Authored: Tue Aug 2 22:32:21 2016 -0700
Committer: Chris Nauroth <cn...@apache.org>
Committed: Tue Aug 2 22:32:21 2016 -0700
----------------------------------------------------------------------
.../hadoop/fs/azure/AbstractWasbTestBase.java | 66 ++++++++++++++++++++
.../fs/azure/NativeAzureFileSystemBaseTest.java | 28 +--------
...estFileSystemOperationExceptionHandling.java | 4 +-
...perationsExceptionHandlingMultiThreaded.java | 6 +-
.../azure/TestNativeAzureFileSystemAppend.java | 2 +-
...ativeAzureFileSystemAtomicRenameDirList.java | 3 +-
.../TestNativeAzureFileSystemClientLogging.java | 6 +-
7 files changed, 79 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ef1bc0b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/AbstractWasbTestBase.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/AbstractWasbTestBase.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/AbstractWasbTestBase.java
new file mode 100644
index 0000000..f0df38b
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/AbstractWasbTestBase.java
@@ -0,0 +1,66 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azure;
+
+import static org.junit.Assume.assumeNotNull;
+
+import org.apache.hadoop.fs.FileSystem;
+
+import org.junit.After;
+import org.junit.Before;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Abstract test class that provides basic setup and teardown of testing Azure
+ * Storage account. Each subclass defines a different set of test cases to run
+ * and overrides {@link #createTestAccount()} to set up the testing account used
+ * to run those tests. The returned account might integrate with Azure Storage
+ * directly or it might be a mock implementation.
+ */
+abstract class AbstractWasbTestBase {
+
+ protected static final Logger LOG =
+ LoggerFactory.getLogger(AbstractWasbTestBase.class);
+
+ protected FileSystem fs;
+ private AzureBlobStorageTestAccount testAccount;
+
+ @Before
+ public void setUp() throws Exception {
+ testAccount = createTestAccount();
+ if (testAccount != null) {
+ fs = testAccount.getFileSystem();
+ }
+ assumeNotNull(testAccount);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (testAccount != null) {
+ testAccount.cleanup();
+ testAccount = null;
+ fs = null;
+ }
+ }
+
+ protected abstract AzureBlobStorageTestAccount createTestAccount()
+ throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ef1bc0b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/NativeAzureFileSystemBaseTest.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/NativeAzureFileSystemBaseTest.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/NativeAzureFileSystemBaseTest.java
index 60e5ec6..177477c 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/NativeAzureFileSystemBaseTest.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/NativeAzureFileSystemBaseTest.java
@@ -23,7 +23,6 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import static org.junit.Assume.assumeNotNull;
import java.io.BufferedReader;
import java.io.BufferedWriter;
@@ -50,8 +49,6 @@ import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.security.UserGroupInformation;
-import org.junit.After;
-import org.junit.Before;
import org.junit.Test;
import org.apache.hadoop.fs.azure.AzureException;
import org.apache.hadoop.fs.azure.NativeAzureFileSystem.FolderRenamePending;
@@ -68,34 +65,13 @@ import com.microsoft.azure.storage.blob.CloudBlob;
* For hand-testing: remove "abstract" keyword and copy in an implementation of createTestAccount
* from one of the subclasses
*/
-public abstract class NativeAzureFileSystemBaseTest {
+public abstract class NativeAzureFileSystemBaseTest
+ extends AbstractWasbTestBase {
- protected FileSystem fs;
- private AzureBlobStorageTestAccount testAccount;
private final long modifiedTimeErrorMargin = 5 * 1000; // Give it +/-5 seconds
- protected abstract AzureBlobStorageTestAccount createTestAccount() throws Exception;
-
public static final Log LOG = LogFactory.getLog(NativeAzureFileSystemBaseTest.class);
- @Before
- public void setUp() throws Exception {
- testAccount = createTestAccount();
- if (testAccount != null) {
- fs = testAccount.getFileSystem();
- }
- assumeNotNull(testAccount);
- }
-
- @After
- public void tearDown() throws Exception {
- if (testAccount != null) {
- testAccount.cleanup();
- testAccount = null;
- fs = null;
- }
- }
-
@Test
public void testCheckingNonExistentOneLetterFile() throws Exception {
assertFalse(fs.exists(new Path("/a")));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ef1bc0b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationExceptionHandling.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationExceptionHandling.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationExceptionHandling.java
index e32c692..f5a1754 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationExceptionHandling.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationExceptionHandling.java
@@ -31,8 +31,8 @@ import org.junit.Assert;
import org.junit.Test;
-public class TestFileSystemOperationExceptionHandling extends
- NativeAzureFileSystemBaseTest {
+public class TestFileSystemOperationExceptionHandling
+ extends AbstractWasbTestBase {
private FSDataInputStream inputStream = null;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ef1bc0b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationsExceptionHandlingMultiThreaded.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationsExceptionHandlingMultiThreaded.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationsExceptionHandlingMultiThreaded.java
index 35d528a..4c56795 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationsExceptionHandlingMultiThreaded.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationsExceptionHandlingMultiThreaded.java
@@ -29,8 +29,8 @@ import org.apache.hadoop.fs.permission.FsPermission;
import org.junit.After;
import org.junit.Test;
-public class TestFileSystemOperationsExceptionHandlingMultiThreaded extends
- NativeAzureFileSystemBaseTest {
+public class TestFileSystemOperationsExceptionHandlingMultiThreaded
+ extends AbstractWasbTestBase {
FSDataInputStream inputStream = null;
@@ -326,4 +326,4 @@ class DeleteThread implements Runnable {
// by the other thread
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ef1bc0b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAppend.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAppend.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAppend.java
index de51990..a2b35cb 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAppend.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAppend.java
@@ -32,7 +32,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
-public class TestNativeAzureFileSystemAppend extends NativeAzureFileSystemBaseTest {
+public class TestNativeAzureFileSystemAppend extends AbstractWasbTestBase {
private static final String TEST_FILE = "test.dat";
private static final Path TEST_PATH = new Path(TEST_FILE);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ef1bc0b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAtomicRenameDirList.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAtomicRenameDirList.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAtomicRenameDirList.java
index b9cca25..602c1f7 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAtomicRenameDirList.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAtomicRenameDirList.java
@@ -24,7 +24,8 @@ import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.junit.Test;
-public class TestNativeAzureFileSystemAtomicRenameDirList extends NativeAzureFileSystemBaseTest {
+public class TestNativeAzureFileSystemAtomicRenameDirList
+ extends AbstractWasbTestBase {
private AzureBlobStorageTestAccount testAccount;
// HBase-site config controlling HBase root dir
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ef1bc0b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemClientLogging.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemClientLogging.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemClientLogging.java
index da39fa3..c9cce75 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemClientLogging.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemClientLogging.java
@@ -37,8 +37,8 @@ import org.junit.Test;
* client-side logging.
*
*/
-public class TestNativeAzureFileSystemClientLogging extends
- NativeAzureFileSystemBaseTest {
+public class TestNativeAzureFileSystemClientLogging
+ extends AbstractWasbTestBase {
private AzureBlobStorageTestAccount testAccount;
@@ -127,4 +127,4 @@ public class TestNativeAzureFileSystemClientLogging extends
testAccount = AzureBlobStorageTestAccount.create();
return testAccount;
}
-}
\ No newline at end of file
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[02/30] hadoop git commit: MAPREDUCE-6724. Single shuffle to memory
must not exceed Integer#MAX_VALUE. (Haibo Chen via gera)
Posted by dr...@apache.org.
MAPREDUCE-6724. Single shuffle to memory must not exceed Integer#MAX_VALUE. (Haibo Chen via gera)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/6890d5b4
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/6890d5b4
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/6890d5b4
Branch: refs/heads/HADOOP-12756
Commit: 6890d5b472320fa7592ed1b08b623c55a27089c6
Parents: c4463f2
Author: Gera Shegalov <ge...@apache.org>
Authored: Thu Jul 28 14:37:03 2016 -0700
Committer: Gera Shegalov <ge...@apache.org>
Committed: Mon Aug 1 23:35:47 2016 -0700
----------------------------------------------------------------------
.../mapreduce/task/reduce/MergeManagerImpl.java | 26 +++++++++++---------
.../mapreduce/task/reduce/TestMergeManager.java | 23 +++++++++++------
2 files changed, 30 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/6890d5b4/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java
index 1673ff8..09fe0cb 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java
@@ -99,7 +99,9 @@ public class MergeManagerImpl<K, V> implements MergeManager<K, V> {
private long usedMemory;
private long commitMemory;
- private final long maxSingleShuffleLimit;
+
+ @VisibleForTesting
+ final long maxSingleShuffleLimit;
private final int memToMemMergeOutputsThreshold;
private final long mergeThreshold;
@@ -187,10 +189,16 @@ public class MergeManagerImpl<K, V> implements MergeManager<K, V> {
usedMemory = 0L;
commitMemory = 0L;
- this.maxSingleShuffleLimit =
- (long)(memoryLimit * singleShuffleMemoryLimitPercent);
- this.memToMemMergeOutputsThreshold =
- jobConf.getInt(MRJobConfig.REDUCE_MEMTOMEM_THRESHOLD, ioSortFactor);
+ long maxSingleShuffleLimitConfiged =
+ (long)(memoryLimit * singleShuffleMemoryLimitPercent);
+ if(maxSingleShuffleLimitConfiged > Integer.MAX_VALUE) {
+ maxSingleShuffleLimitConfiged = Integer.MAX_VALUE;
+ LOG.info("The max number of bytes for a single in-memory shuffle cannot" +
+ " be larger than Integer.MAX_VALUE. Setting it to Integer.MAX_VALUE");
+ }
+ this.maxSingleShuffleLimit = maxSingleShuffleLimitConfiged;
+ this.memToMemMergeOutputsThreshold =
+ jobConf.getInt(MRJobConfig.REDUCE_MEMTOMEM_THRESHOLD, ioSortFactor);
this.mergeThreshold = (long)(this.memoryLimit *
jobConf.getFloat(
MRJobConfig.SHUFFLE_MERGE_PERCENT,
@@ -249,17 +257,13 @@ public class MergeManagerImpl<K, V> implements MergeManager<K, V> {
public void waitForResource() throws InterruptedException {
inMemoryMerger.waitForMerge();
}
-
- private boolean canShuffleToMemory(long requestedSize) {
- return (requestedSize < maxSingleShuffleLimit);
- }
-
+
@Override
public synchronized MapOutput<K,V> reserve(TaskAttemptID mapId,
long requestedSize,
int fetcher
) throws IOException {
- if (!canShuffleToMemory(requestedSize)) {
+ if (requestedSize > maxSingleShuffleLimit) {
LOG.info(mapId + ": Shuffling to disk since " + requestedSize +
" is greater than maxSingleShuffleLimit (" +
maxSingleShuffleLimit + ")");
http://git-wip-us.apache.org/repos/asf/hadoop/blob/6890d5b4/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMergeManager.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMergeManager.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMergeManager.java
index 1c0d25b..325d2f9 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMergeManager.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMergeManager.java
@@ -289,22 +289,29 @@ public class TestMergeManager {
final long maxInMemReduce = mgr.getMaxInMemReduceLimit();
assertTrue("Large in-memory reduce area unusable: " + maxInMemReduce,
maxInMemReduce > Integer.MAX_VALUE);
+ assertEquals("maxSingleShuffleLimit to be capped at Integer.MAX_VALUE",
+ Integer.MAX_VALUE, mgr.maxSingleShuffleLimit);
+ verifyReservedMapOutputType(mgr, 10L, "MEMORY");
+ verifyReservedMapOutputType(mgr, 1L + Integer.MAX_VALUE, "DISK");
+ }
+
+ private void verifyReservedMapOutputType(MergeManagerImpl<Text, Text> mgr,
+ long size, String expectedShuffleMode) throws IOException {
+ final TaskAttemptID mapId = TaskAttemptID.forName("attempt_0_1_m_1_1");
+ final MapOutput<Text, Text> mapOutput = mgr.reserve(mapId, size, 1);
+ assertEquals("Shuffled bytes: " + size, expectedShuffleMode,
+ mapOutput.getDescription());
+ mgr.unreserve(size);
}
@Test
public void testZeroShuffleMemoryLimitPercent() throws Exception {
final JobConf jobConf = new JobConf();
jobConf.setFloat(MRJobConfig.SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f);
- final MergeManager<Text, Text> mgr =
+ final MergeManagerImpl<Text, Text> mgr =
new MergeManagerImpl<>(null, jobConf, mock(LocalFileSystem.class),
null, null, null, null, null, null, null, null, null, null,
new MROutputFiles());
- final long mapOutputSize = 10;
- final int fetcher = 1;
- final MapOutput<Text, Text> mapOutput = mgr.reserve(
- TaskAttemptID.forName("attempt_0_1_m_1_1"),
- mapOutputSize, fetcher);
- assertEquals("Tiny map outputs should be shuffled to disk", "DISK",
- mapOutput.getDescription());
+ verifyReservedMapOutputType(mgr, 10L, "DISK");
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[29/30] hadoop git commit: YARN-5459. Add support for docker rm.
Contributed by Shane Kumpf.
Posted by dr...@apache.org.
YARN-5459. Add support for docker rm. Contributed by Shane Kumpf.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/08e33381
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/08e33381
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/08e33381
Branch: refs/heads/HADOOP-12756
Commit: 08e33381f5342590a68c65504c79c296b40fe5af
Parents: 05db649
Author: Varun Vasudev <vv...@apache.org>
Authored: Thu Aug 4 14:08:34 2016 +0530
Committer: Varun Vasudev <vv...@apache.org>
Committed: Thu Aug 4 14:08:34 2016 +0530
----------------------------------------------------------------------
.../linux/runtime/docker/DockerRmCommand.java | 30 ++++++++++++
.../runtime/docker/TestDockerRmCommand.java | 48 ++++++++++++++++++++
2 files changed, 78 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/08e33381/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerRmCommand.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerRmCommand.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerRmCommand.java
new file mode 100644
index 0000000..b1aea61
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerRmCommand.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker;
+
+/**
+ * Encapsulates the docker rm command and its command
+ * line arguments.
+ */
+public class DockerRmCommand extends DockerCommand {
+ private static final String RM_COMMAND = "rm";
+
+ public DockerRmCommand(String containerName) {
+ super(RM_COMMAND);
+ super.addCommandArguments(containerName);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/08e33381/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerRmCommand.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerRmCommand.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerRmCommand.java
new file mode 100644
index 0000000..d1b9904
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerRmCommand.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests the docker rm command and any command
+ * line arguments.
+ */
+public class TestDockerRmCommand {
+
+ private DockerRmCommand dockerRmCommand;
+
+ private static final String CONTAINER_NAME = "foo";
+
+ @Before
+ public void setUp() {
+ dockerRmCommand = new DockerRmCommand(CONTAINER_NAME);
+ }
+
+ @Test
+ public void testGetCommandOption() {
+ assertEquals("rm", dockerRmCommand.getCommandOption());
+ }
+
+ @Test
+ public void testGetCommandWithArguments() {
+ assertEquals("rm foo", dockerRmCommand.getCommandWithArguments());
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[14/30] hadoop git commit: YARN-5460. Change container runtime type
logging in DelegatingLinuxContainerRuntime to debug. Contributed by Shane
Kumpf.
Posted by dr...@apache.org.
YARN-5460. Change container runtime type logging in DelegatingLinuxContainerRuntime to debug. Contributed by Shane Kumpf.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/d848184e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/d848184e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/d848184e
Branch: refs/heads/HADOOP-12756
Commit: d848184e905d70835293ceac7a01b5a0ca21cc0a
Parents: 4ef1bc0
Author: Varun Vasudev <vv...@apache.org>
Authored: Wed Aug 3 12:02:39 2016 +0530
Committer: Varun Vasudev <vv...@apache.org>
Committed: Wed Aug 3 12:02:39 2016 +0530
----------------------------------------------------------------------
.../linux/runtime/DelegatingLinuxContainerRuntime.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/d848184e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DelegatingLinuxContainerRuntime.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DelegatingLinuxContainerRuntime.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DelegatingLinuxContainerRuntime.java
index 75abfb0..975bb9b 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DelegatingLinuxContainerRuntime.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DelegatingLinuxContainerRuntime.java
@@ -63,8 +63,8 @@ public class DelegatingLinuxContainerRuntime implements LinuxContainerRuntime {
runtime = defaultLinuxContainerRuntime;
}
- if (LOG.isInfoEnabled()) {
- LOG.info("Using container runtime: " + runtime.getClass()
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Using container runtime: " + runtime.getClass()
.getSimpleName());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org