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 sa...@apache.org on 2019/07/19 07:40:05 UTC
[hadoop] 01/05: HDDS-1713. ReplicationManager fail to find proper
node topology based Datanode details from heartbeat
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1713
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit 6e588145224ec6742e7ac07625b049ea9d414e53
Author: Sammi Chen <sa...@apache.org>
AuthorDate: Wed Jul 17 21:35:41 2019 +0800
HDDS-1713. ReplicationManager fail to find proper node topology based Datanode details from heartbeat
---
.../hadoop/hdds/protocol/DatanodeDetails.java | 15 ++-
.../states/endpoint/RegisterEndpointTask.java | 4 +
.../ozone/protocol/commands/RegisteredCommand.java | 121 ++++++---------------
.../proto/StorageContainerDatanodeProtocol.proto | 2 +
.../apache/hadoop/hdds/scm/node/NodeManager.java | 16 ++-
.../hadoop/hdds/scm/node/SCMNodeManager.java | 81 ++++++++------
.../hdds/scm/pipeline/PipelineActionHandler.java | 2 +-
.../hdds/scm/server/SCMBlockProtocolServer.java | 8 +-
.../hdds/scm/server/SCMDatanodeProtocolServer.java | 10 +-
.../java/org/apache/hadoop/hdds/scm/TestUtils.java | 6 +-
.../hadoop/hdds/scm/container/MockNodeManager.java | 9 +-
.../TestSCMContainerPlacementRackAware.java | 4 -
.../hadoop/hdds/scm/node/TestSCMNodeManager.java | 4 +-
.../testutils/ReplicationNodeManagerMock.java | 7 +-
.../hadoop/ozone/TestStorageContainerManager.java | 67 ++++++++++++
.../org/apache/hadoop/ozone/om/KeyManagerImpl.java | 7 +-
16 files changed, 210 insertions(+), 153 deletions(-)
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java
index 31e4df0..698a443 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.hdds.protocol;
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
@@ -193,12 +194,12 @@ public class DatanodeDetails extends NodeImpl implements
builder.addPort(newPort(
Port.Name.valueOf(port.getName().toUpperCase()), port.getValue()));
}
- if (datanodeDetailsProto.hasNetworkLocation()) {
- builder.setNetworkLocation(datanodeDetailsProto.getNetworkLocation());
- }
if (datanodeDetailsProto.hasNetworkName()) {
builder.setNetworkName(datanodeDetailsProto.getNetworkName());
}
+ if (datanodeDetailsProto.hasNetworkLocation()) {
+ builder.setNetworkLocation(datanodeDetailsProto.getNetworkLocation());
+ }
return builder.build();
}
@@ -219,8 +220,12 @@ public class DatanodeDetails extends NodeImpl implements
if (certSerialId != null) {
builder.setCertSerialId(certSerialId);
}
- builder.setNetworkLocation(getNetworkLocation());
- builder.setNetworkName(getNetworkName());
+ if (!Strings.isNullOrEmpty(getNetworkName())) {
+ builder.setNetworkName(getNetworkName());
+ }
+ if (!Strings.isNullOrEmpty(getNetworkLocation())) {
+ builder.setNetworkLocation(getNetworkLocation());
+ }
for (Port port : ports) {
builder.addPorts(HddsProtos.Port.newBuilder()
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/RegisterEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/RegisterEndpointTask.java
index 9918f9d..b94b1cf 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/RegisterEndpointTask.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/RegisterEndpointTask.java
@@ -128,6 +128,10 @@ public final class RegisterEndpointTask implements
datanodeDetails.setHostName(response.getHostname());
datanodeDetails.setIpAddress(response.getIpAddress());
}
+ if (response.hasNetworkName() && response.hasNetworkLocation()) {
+ datanodeDetails.setNetworkName(response.getNetworkName());
+ datanodeDetails.setNetworkLocation(response.getNetworkLocation());
+ }
EndpointStateMachine.EndPointStates nextState =
rpcEndPoint.getState().getNextState();
rpcEndPoint.setState(nextState);
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/RegisteredCommand.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/RegisteredCommand.java
index 3a5da72..42778cb 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/RegisteredCommand.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/RegisteredCommand.java
@@ -17,7 +17,8 @@
*/
package org.apache.hadoop.ozone.protocol.commands;
-import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.SCMRegisteredResponseProto;
import org.apache.hadoop.hdds.protocol.proto
@@ -28,23 +29,15 @@ import org.apache.hadoop.hdds.protocol.proto
* Response to Datanode Register call.
*/
public class RegisteredCommand {
- private String datanodeUUID;
private String clusterID;
private ErrorCode error;
- private String hostname;
- private String ipAddress;
+ private DatanodeDetails datanode;
- public RegisteredCommand(final ErrorCode error, final String datanodeUUID,
+ public RegisteredCommand(final ErrorCode error, final DatanodeDetails node,
final String clusterID) {
- this(error, datanodeUUID, clusterID, null, null);
- }
- public RegisteredCommand(final ErrorCode error, final String datanodeUUID,
- final String clusterID, final String hostname, final String ipAddress) {
- this.datanodeUUID = datanodeUUID;
+ this.datanode = node;
this.clusterID = clusterID;
this.error = error;
- this.hostname = hostname;
- this.ipAddress = ipAddress;
}
/**
@@ -57,12 +50,12 @@ public class RegisteredCommand {
}
/**
- * Returns datanode UUID.
+ * Returns datanode.
*
- * @return - Datanode ID.
+ * @return - Datanode.
*/
- public String getDatanodeUUID() {
- return datanodeUUID;
+ public DatanodeDetails getDatanode() {
+ return datanode;
}
/**
@@ -84,79 +77,54 @@ public class RegisteredCommand {
}
/**
- * Returns the hostname.
- *
- * @return - hostname
- */
- public String getHostName() {
- return hostname;
- }
-
- /**
- * Returns the ipAddress of the dataNode.
- */
- public String getIpAddress() {
- return ipAddress;
- }
-
- /**
* Gets the protobuf message of this object.
*
* @return A protobuf message.
*/
- public byte[] getProtoBufMessage() {
+ public SCMRegisteredResponseProto getProtoBufMessage() {
SCMRegisteredResponseProto.Builder builder =
SCMRegisteredResponseProto.newBuilder()
+ // TODO : Fix this later when we have multiple SCM support.
+ // .setAddressList(addressList)
.setClusterID(this.clusterID)
- .setDatanodeUUID(this.datanodeUUID)
+ .setDatanodeUUID(this.datanode.getUuidString())
.setErrorCode(this.error);
- if (hostname != null && ipAddress != null) {
- builder.setHostname(hostname).setIpAddress(ipAddress);
+ if (!Strings.isNullOrEmpty(datanode.getHostName())) {
+ builder.setHostname(datanode.getHostName());
+ }
+ if (!Strings.isNullOrEmpty(datanode.getIpAddress())) {
+ builder.setIpAddress(datanode.getIpAddress());
+ }
+ if (!Strings.isNullOrEmpty(datanode.getNetworkName())) {
+ builder.setNetworkName(datanode.getNetworkName());
}
- return builder.build().toByteArray();
+ if (!Strings.isNullOrEmpty(datanode.getNetworkLocation())) {
+ builder.setNetworkLocation(datanode.getNetworkLocation());
+ }
+
+ return builder.build();
}
/**
* A builder class to verify all values are sane.
*/
public static class Builder {
- private String datanodeUUID;
+ private DatanodeDetails datanode;
private String clusterID;
private ErrorCode error;
- private String ipAddress;
- private String hostname;
/**
- * sets UUID.
+ * sets datanode details.
*
- * @param dnUUID - datanode UUID
+ * @param node - datanode details
* @return Builder
*/
- public Builder setDatanodeUUID(String dnUUID) {
- this.datanodeUUID = dnUUID;
+ public Builder setDatanode(DatanodeDetails node) {
+ this.datanode = node;
return this;
}
/**
- * Create this object from a Protobuf message.
- *
- * @param response - RegisteredCmdResponseProto
- * @return RegisteredCommand
- */
- public RegisteredCommand getFromProtobuf(SCMRegisteredResponseProto
- response) {
- Preconditions.checkNotNull(response);
- if (response.hasHostname() && response.hasIpAddress()) {
- return new RegisteredCommand(response.getErrorCode(),
- response.getDatanodeUUID(), response.getClusterID(),
- response.getHostname(), response.getIpAddress());
- } else {
- return new RegisteredCommand(response.getErrorCode(),
- response.getDatanodeUUID(), response.getClusterID());
- }
- }
-
- /**
* Sets cluster ID.
*
* @param cluster - clusterID
@@ -179,37 +147,18 @@ public class RegisteredCommand {
}
/**
- * sets the hostname.
- */
- public Builder setHostname(String host) {
- this.hostname = host;
- return this;
- }
-
- public Builder setIpAddress(String ipAddr) {
- this.ipAddress = ipAddr;
- return this;
- }
-
- /**
* Build the command object.
*
* @return RegisteredCommand
*/
public RegisteredCommand build() {
- if ((this.error == ErrorCode.success) && (this.datanodeUUID == null
- || this.datanodeUUID.isEmpty()) || (this.clusterID == null
- || this.clusterID.isEmpty())) {
+ if ((this.error == ErrorCode.success) && (this.datanode == null
+ || Strings.isNullOrEmpty(this.datanode.getUuidString())
+ || Strings.isNullOrEmpty(this.clusterID))) {
throw new IllegalArgumentException("On success, RegisteredCommand "
+ "needs datanodeUUID and ClusterID.");
}
- if (hostname != null && ipAddress != null) {
- return new RegisteredCommand(this.error, this.datanodeUUID,
- this.clusterID, this.hostname, this.ipAddress);
- } else {
- return new RegisteredCommand(this.error, this.datanodeUUID,
- this.clusterID);
- }
+ return new RegisteredCommand(this.error, this.datanode, this.clusterID);
}
}
}
diff --git a/hadoop-hdds/container-service/src/main/proto/StorageContainerDatanodeProtocol.proto b/hadoop-hdds/container-service/src/main/proto/StorageContainerDatanodeProtocol.proto
index 73dc1cc..500735a 100644
--- a/hadoop-hdds/container-service/src/main/proto/StorageContainerDatanodeProtocol.proto
+++ b/hadoop-hdds/container-service/src/main/proto/StorageContainerDatanodeProtocol.proto
@@ -70,6 +70,8 @@ message SCMRegisteredResponseProto {
optional SCMNodeAddressList addressList = 4;
optional string hostname = 5;
optional string ipAddress = 6;
+ optional string networkName = 7;
+ optional string networkLocation = 8;
}
/**
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
index 6b05772..0ccbb82 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
@@ -173,11 +173,19 @@ public interface NodeManager extends StorageContainerNodeProtocol,
List<SCMCommand> getCommandQueue(UUID dnID);
/**
- * Given datanode host address, returns the DatanodeDetails for the
- * node.
+ * Given datanode uuid, returns the DatanodeDetails for the node.
*
- * @param address node host address
+ * @param uuid datanode uuid
* @return the given datanode, or null if not found
*/
- DatanodeDetails getNode(String address);
+ DatanodeDetails getNodeByUuid(String uuid);
+
+ /**
+ * Given datanode address(Ipaddress or hostname), returns the DatanodeDetails
+ * for the node.
+ *
+ * @param address datanode address
+ * @return the given datanode, or null if not found
+ */
+ DatanodeDetails getNodeByAddress(String address);
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
index e4e9d35..93ca304 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
@@ -19,15 +19,13 @@ package org.apache.hadoop.hdds.scm.node;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import org.apache.hadoop.hdds.scm.container.ContainerID;
-import org.apache.hadoop.hdds.scm.net.InnerNode;
-import org.apache.hadoop.hdds.scm.net.NetConstants;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
-import org.apache.hadoop.hdds.scm.net.Node;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.node.states.NodeAlreadyExistsException;
@@ -74,6 +72,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.stream.Collectors;
@@ -106,6 +105,8 @@ public class SCMNodeManager implements NodeManager {
private final NetworkTopology clusterMap;
private final DNSToSwitchMapping dnsToSwitchMapping;
private final boolean useHostname;
+ private final ConcurrentHashMap<String, String> dnsToUuidMap =
+ new ConcurrentHashMap<>();
/**
* Constructs SCM machine Manager.
@@ -252,19 +253,21 @@ public class SCMNodeManager implements NodeManager {
datanodeDetails.setIpAddress(dnAddress.getHostAddress());
}
try {
- String location;
+ String dnsName;
+ String networkLocation;
+ datanodeDetails.setNetworkName(datanodeDetails.getUuidString());
if (useHostname) {
- datanodeDetails.setNetworkName(datanodeDetails.getHostName());
- location = nodeResolve(datanodeDetails.getHostName());
+ dnsName = datanodeDetails.getHostName();
} else {
- datanodeDetails.setNetworkName(datanodeDetails.getIpAddress());
- location = nodeResolve(datanodeDetails.getIpAddress());
+ dnsName = datanodeDetails.getIpAddress();
}
- if (location != null) {
- datanodeDetails.setNetworkLocation(location);
+ networkLocation = nodeResolve(dnsName);
+ if (networkLocation != null) {
+ datanodeDetails.setNetworkLocation(networkLocation);
}
nodeStateManager.addNode(datanodeDetails);
clusterMap.add(datanodeDetails);
+ dnsToUuidMap.put(dnsName, datanodeDetails.getUuidString());
// Updating Node Report, as registration is successful
processNodeReport(datanodeDetails, nodeReport);
LOG.info("Registered Data node : {}", datanodeDetails);
@@ -274,10 +277,8 @@ public class SCMNodeManager implements NodeManager {
}
return RegisteredCommand.newBuilder().setErrorCode(ErrorCode.success)
- .setDatanodeUUID(datanodeDetails.getUuidString())
+ .setDatanode(datanodeDetails)
.setClusterID(this.scmStorageConfig.getClusterID())
- .setHostname(datanodeDetails.getHostName())
- .setIpAddress(datanodeDetails.getIpAddress())
.build();
}
@@ -553,33 +554,49 @@ public class SCMNodeManager implements NodeManager {
}
/**
- * Given datanode address or host name, returns the DatanodeDetails for the
- * node.
+ * Given datanode uuid, returns the DatanodeDetails for the node.
*
- * @param address node host address
+ * @param uuid node host address
* @return the given datanode, or null if not found
*/
@Override
- public DatanodeDetails getNode(String address) {
- Node node = null;
- String location = nodeResolve(address);
- if (location != null) {
- node = clusterMap.getNode(location + NetConstants.PATH_SEPARATOR_STR +
- address);
+ public DatanodeDetails getNodeByUuid(String uuid) {
+ if (Strings.isNullOrEmpty(uuid)) {
+ LOG.warn("uuid is null");
+ return null;
+ }
+ DatanodeDetails temp = DatanodeDetails.newBuilder().setUuid(uuid).build();
+ try {
+ return nodeStateManager.getNode(temp);
+ } catch (NodeNotFoundException e) {
+ LOG.warn("Cannot find node for uuid {}", uuid);
+ return null;
}
+ }
- if (node != null) {
- if (node instanceof InnerNode) {
- LOG.warn("Get node for {} return {}, it's an inner node, " +
- "not a datanode", address, node.getNetworkFullPath());
- } else {
- LOG.debug("Get node for {} return {}", address,
- node.getNetworkFullPath());
- return (DatanodeDetails)node;
+ /**
+ * Given datanode address(Ipaddress or hostname), returns the DatanodeDetails
+ * for the node.
+ *
+ * @param address datanode address
+ * @return the given datanode, or null if not found
+ */
+ @Override
+ public DatanodeDetails getNodeByAddress(String address) {
+ if (Strings.isNullOrEmpty(address)) {
+ LOG.warn("address is null");
+ return null;
+ }
+ String uuid = dnsToUuidMap.get(address);
+ if (uuid != null) {
+ DatanodeDetails temp = DatanodeDetails.newBuilder().setUuid(uuid).build();
+ try {
+ return nodeStateManager.getNode(temp);
+ } catch (NodeNotFoundException e) {
+ LOG.warn("Cannot find node for uuid {}", uuid);
}
- } else {
- LOG.warn("Cannot find node for {}", address);
}
+ LOG.warn("Cannot find node for address {}", address);
return null;
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineActionHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineActionHandler.java
index da704d2..955bfc6 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineActionHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineActionHandler.java
@@ -57,7 +57,7 @@ public class PipelineActionHandler
pipelineID = PipelineID.
getFromProtobuf(action.getClosePipeline().getPipelineID());
Pipeline pipeline = pipelineManager.getPipeline(pipelineID);
- LOG.info("Received pipeline action {} for {} from datanode [}",
+ LOG.info("Received pipeline action {} for {} from datanode {}",
action.getAction(), pipeline, report.getDatanodeDetails());
pipelineManager.finalizeAndDestroyPipeline(pipeline, true);
} catch (IOException ioe) {
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMBlockProtocolServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMBlockProtocolServer.java
index 1c76070..35ec295 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMBlockProtocolServer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMBlockProtocolServer.java
@@ -288,12 +288,12 @@ public class SCMBlockProtocolServer implements
boolean auditSuccess = true;
try{
NodeManager nodeManager = scm.getScmNodeManager();
- Node client = nodeManager.getNode(clientMachine);
+ Node client = nodeManager.getNodeByAddress(clientMachine);
List<Node> nodeList = new ArrayList();
- nodes.stream().forEach(path -> {
- DatanodeDetails node = nodeManager.getNode(path);
+ nodes.stream().forEach(uuid -> {
+ DatanodeDetails node = nodeManager.getNodeByUuid(uuid);
if (node != null) {
- nodeList.add(nodeManager.getNode(path));
+ nodeList.add(node);
}
});
List<? extends Node> sortedNodeList = scm.getClusterMap()
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
index cd78d3d..6dd9dab 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
@@ -265,15 +265,7 @@ public class SCMDatanodeProtocolServer implements
@VisibleForTesting
public static SCMRegisteredResponseProto getRegisteredResponse(
RegisteredCommand cmd) {
- return SCMRegisteredResponseProto.newBuilder()
- // TODO : Fix this later when we have multiple SCM support.
- // .setAddressList(addressList)
- .setErrorCode(cmd.getError())
- .setClusterID(cmd.getClusterID())
- .setDatanodeUUID(cmd.getDatanodeUUID())
- .setIpAddress(cmd.getIpAddress())
- .setHostname(cmd.getHostName())
- .build();
+ return cmd.getProtoBufMessage();
}
@Override
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestUtils.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestUtils.java
index e2ce7de..37321d7 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestUtils.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestUtils.java
@@ -137,8 +137,10 @@ public final class TestUtils {
*/
public static DatanodeDetails getDatanodeDetails(
RegisteredCommand registeredCommand) {
- return createDatanodeDetails(registeredCommand.getDatanodeUUID(),
- registeredCommand.getHostName(), registeredCommand.getIpAddress(),
+ return createDatanodeDetails(
+ registeredCommand.getDatanode().getUuidString(),
+ registeredCommand.getDatanode().getHostName(),
+ registeredCommand.getDatanode().getIpAddress(),
null);
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java
index 64eba29..c7e26ad 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java
@@ -459,11 +459,16 @@ public class MockNodeManager implements NodeManager {
}
@Override
- public DatanodeDetails getNode(String address) {
- Node node = clusterMap.getNode(NetConstants.DEFAULT_RACK + "/" + address);
+ public DatanodeDetails getNodeByUuid(String uuid) {
+ Node node = clusterMap.getNode(NetConstants.DEFAULT_RACK + "/" + uuid);
return node == null ? null : (DatanodeDetails)node;
}
+ @Override
+ public DatanodeDetails getNodeByAddress(String address) {
+ return null;
+ }
+
public void setNetworkTopology(NetworkTopology topology) {
this.clusterMap = topology;
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java
index e63b09e..4c68721 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java
@@ -137,10 +137,6 @@ public class TestSCMContainerPlacementRackAware {
datanodeDetails.get(2)));
Assert.assertFalse(cluster.isSameParent(datanodeDetails.get(1),
datanodeDetails.get(2)));
- Assert.assertFalse(cluster.isSameParent(datanodeDetails.get(0),
- datanodeDetails.get(3)));
- Assert.assertFalse(cluster.isSameParent(datanodeDetails.get(2),
- datanodeDetails.get(3)));
}
@Test
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java
index 4657fa0..d028851 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java
@@ -1152,10 +1152,10 @@ public class TestSCMNodeManager {
// test get node
if (useHostname) {
Arrays.stream(hostNames).forEach(hostname ->
- Assert.assertNotNull(nodeManager.getNode(hostname)));
+ Assert.assertNotNull(nodeManager.getNodeByAddress(hostname)));
} else {
Arrays.stream(ipAddress).forEach(ip ->
- Assert.assertNotNull(nodeManager.getNode(ip)));
+ Assert.assertNotNull(nodeManager.getNodeByAddress(ip)));
}
}
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java
index bc26e3c..aa5dc2f 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java
@@ -311,7 +311,12 @@ public class ReplicationNodeManagerMock implements NodeManager {
}
@Override
- public DatanodeDetails getNode(String address) {
+ public DatanodeDetails getNodeByUuid(String address) {
+ return null;
+ }
+
+ @Override
+ public DatanodeDetails getNodeByAddress(String address) {
return null;
}
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManager.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManager.java
index 3eb1ebd..fc91682 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManager.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManager.java
@@ -17,6 +17,10 @@
*/
package org.apache.hadoop.ozone;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic
+ .NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic
+ .NET_TOPOLOGY_TABLE_MAPPING_FILE_KEY;
import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_COMMAND_STATUS_REPORT_INTERVAL;
import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_REPORT_INTERVAL;
import static org.junit.Assert.fail;
@@ -26,6 +30,8 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.io.BufferedWriter;
+import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
@@ -39,9 +45,13 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeType;
@@ -58,6 +68,7 @@ import org.apache.hadoop.hdds.scm.container.ReplicationManager;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
import org.apache.hadoop.hdds.scm.node.NodeManager;
import org.apache.hadoop.hdds.scm.server.SCMClientProtocolServer;
import org.apache.hadoop.hdds.scm.server.SCMStorageConfig;
@@ -72,6 +83,7 @@ import org.apache.hadoop.ozone.protocol.commands.DeleteBlocksCommand;
import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
import org.apache.hadoop.security.authentication.client.AuthenticationException;
import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.util.Time;
import org.apache.hadoop.utils.HddsVersionInfo;
import org.junit.Assert;
import org.junit.Rule;
@@ -483,6 +495,61 @@ public class TestStorageContainerManager {
Assert.assertEquals(expectedVersion, actualVersion);
}
+ /**
+ * Test datanode heartbeat well processed with a 4-layer network topology.
+ */
+ @Test(timeout = 30000)
+ public void testScmProcessDatanodeHeartbeat() throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ String scmId = UUID.randomUUID().toString();
+ String hostname = HddsUtils.getHostName(conf);
+
+ File mapFile = File.createTempFile("rack-mapping", ".txt");
+ mapFile.deleteOnExit();
+ BufferedWriter writer = Files.newWriter(mapFile, Charsets.UTF_8);
+ writer.write(hostname + " /rack1\n");
+ writer.close();
+
+ conf.set(NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+ "org.apache.hadoop.net.TableMapping");
+ conf.set(NET_TOPOLOGY_TABLE_MAPPING_FILE_KEY, mapFile.getCanonicalPath());
+ System.out.println("temp mapping file = " + mapFile.getCanonicalPath());
+ final int datanodeNum = 3;
+ MiniOzoneCluster cluster = MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(datanodeNum)
+ .setScmId(scmId)
+ .build();
+ cluster.waitForClusterToBeReady();
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+
+ try {
+ // first sleep 10s
+ Thread.sleep(10000);
+ // verify datanode heartbeats are well processed
+ long heartbeatCheckerIntervalMs =
+ MiniOzoneCluster.Builder.DEFAULT_HB_INTERVAL_MS;
+ long start = Time.monotonicNow();
+ Thread.sleep(heartbeatCheckerIntervalMs * 2);
+
+ List<DatanodeDetails> allNodes = scm.getScmNodeManager().getAllNodes();
+ Assert.assertTrue(allNodes.size() == datanodeNum);
+ for (int i = 0; i < allNodes.size(); i++) {
+ DatanodeInfo datanodeInfo = (DatanodeInfo) scm.getScmNodeManager()
+ .getNodeByUuid(allNodes.get(i).getUuidString());
+ Assert.assertTrue((datanodeInfo.getLastHeartbeatTime() - start)
+ >= heartbeatCheckerIntervalMs);
+ Assert.assertTrue(datanodeInfo.getUuidString()
+ .equals(datanodeInfo.getNetworkName()));
+ Assert.assertTrue(datanodeInfo.getNetworkLocation()
+ .equals("/rack1"));
+ }
+ } finally {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+ }
+
@Test
@SuppressWarnings("unchecked")
public void testCloseContainerCommandOnRestart() throws Exception {
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java
index 7db4a80..9b4eac3 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java
@@ -2123,9 +2123,14 @@ public class KeyManagerImpl implements KeyManager {
for (OmKeyLocationInfoGroup key : keyInfo.getKeyLocationVersions()) {
key.getLocationList().forEach(k -> {
List<DatanodeDetails> nodes = k.getPipeline().getNodes();
+ if (nodes == null || nodes.size() == 0) {
+ LOG.warn("Datanodes for pipeline {} is empty",
+ k.getPipeline().getId().toString());
+ return;
+ }
List<String> nodeList = new ArrayList<>();
nodes.stream().forEach(node ->
- nodeList.add(node.getNetworkName()));
+ nodeList.add(node.getUuidString()));
try {
List<DatanodeDetails> sortedNodes = scmClient.getBlockClient()
.sortDatanodes(nodeList, clientMachine);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org