This is an automated email from the ASF dual-hosted git repository.
szetszwo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new b72775aedcf HDDS-15682. Use the original configured host and port to
identify SCM nodes (#10612)
b72775aedcf is described below
commit b72775aedcf40457596a8cafdf524a77f4e3e92f
Author: Tsz-Wo Nicholas Sze <[email protected]>
AuthorDate: Sat Jun 27 05:31:56 2026 -0700
HDDS-15682. Use the original configured host and port to identify SCM nodes
(#10612)
---
.../org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java | 53 ++++++++------
.../apache/hadoop/hdds/scm/net/HostAndPort.java | 85 ++++++++++++++++++++++
.../apache/hadoop/ozone/HddsDatanodeService.java | 16 ++--
.../common/statemachine/DatanodeQueueMetrics.java | 18 ++---
.../common/statemachine/EndpointStateMachine.java | 29 +++-----
.../common/statemachine/SCMConnectionManager.java | 14 ++--
.../common/statemachine/StateContext.java | 48 ++++++------
.../common/states/datanode/InitDatanodeState.java | 15 ++--
.../ozone/container/common/ContainerTestUtils.java | 3 +-
.../statemachine/TestSCMConnectionManager.java | 4 +-
.../common/statemachine/TestStateContext.java | 28 +++----
.../states/endpoint/TestHeartbeatEndpointTask.java | 5 +-
.../common/volume/TestVolumeSetDiskChecks.java | 4 +-
.../apache/hadoop/hdds/utils/HddsServerUtil.java | 29 +++-----
.../apache/hadoop/hdds/scm/TestHddsServerUtil.java | 38 +++++-----
.../hadoop/hdds/scm/TestHddsServerUtils.java | 2 +
.../org/apache/hadoop/ozone/recon/ReconServer.java | 4 +-
.../hadoop/ozone/freon/DatanodeSimulator.java | 8 +-
18 files changed, 238 insertions(+), 165 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java
index dc2393fe4a9..98b138de1ce 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMNodeInfo.java
@@ -39,11 +39,13 @@
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import java.util.OptionalInt;
import net.jcip.annotations.Immutable;
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.conf.ConfigurationException;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.ozone.ha.ConfUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,10 +62,10 @@ public class SCMNodeInfo {
private static final Logger LOG = LoggerFactory.getLogger(SCMNodeInfo.class);
private final String serviceId;
private final String nodeId;
- private final String blockClientAddress;
- private final String scmClientAddress;
- private final String scmSecurityAddress;
- private final String scmDatanodeAddress;
+ private final HostAndPort blockClientAddress;
+ private final HostAndPort scmClientAddress;
+ private final HostAndPort scmSecurityAddress;
+ private final HostAndPort scmDatanodeAddress;
/**
* Build SCM Node information from configuration.
@@ -130,6 +132,9 @@ public static List<SCMNodeInfo>
buildNodeInfo(ConfigurationSource conf) {
String scmClientAddress = getHostNameFromConfigKeys(conf,
OZONE_SCM_CLIENT_ADDRESS_KEY,
OZONE_SCM_NAMES).orElse(null);
+ if (scmClientAddress == null) {
+ throw new ConfigurationException(OZONE_SCM_CLIENT_ADDRESS_KEY + " is
not set");
+ }
String scmBlockClientAddress = getHostNameFromConfigKeys(conf,
OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY).orElse(scmClientAddress);
@@ -162,14 +167,10 @@ public static List<SCMNodeInfo>
buildNodeInfo(ConfigurationSource conf) {
scmNodeInfoList.add(new SCMNodeInfo(scmServiceId,
SCM_DUMMY_NODEID,
- scmBlockClientAddress == null ? null :
- buildAddress(scmBlockClientAddress, scmBlockClientPort),
- scmClientAddress == null ? null :
- buildAddress(scmClientAddress, scmClientPort),
- scmSecurityClientAddress == null ? null :
- buildAddress(scmSecurityClientAddress, scmSecurityPort),
- scmDatanodeAddress == null ? null :
- buildAddress(scmDatanodeAddress, scmDatanodePort)));
+ buildAddress(scmBlockClientAddress, scmBlockClientPort),
+ buildAddress(scmClientAddress, scmClientPort),
+ buildAddress(scmSecurityClientAddress, scmSecurityPort),
+ buildAddress(scmDatanodeAddress, scmDatanodePort)));
return scmNodeInfoList;
@@ -177,8 +178,8 @@ public static List<SCMNodeInfo>
buildNodeInfo(ConfigurationSource conf) {
}
- public static String buildAddress(String address, int port) {
- return address + ':' + port;
+ private static HostAndPort buildAddress(String address, int port) {
+ return new HostAndPort(address, port);
}
public static int getPort(ConfigurationSource conf,
@@ -209,14 +210,14 @@ public static int getPort(ConfigurationSource conf,
* @param scmDatanodeAddress
*/
public SCMNodeInfo(String serviceId, String nodeId,
- String blockClientAddress, String scmClientAddress,
- String scmSecurityAddress, String scmDatanodeAddress) {
+ HostAndPort blockClientAddress, HostAndPort scmClientAddress,
+ HostAndPort scmSecurityAddress, HostAndPort scmDatanodeAddress) {
this.serviceId = serviceId;
this.nodeId = nodeId;
- this.blockClientAddress = blockClientAddress;
- this.scmClientAddress = scmClientAddress;
- this.scmSecurityAddress = scmSecurityAddress;
- this.scmDatanodeAddress = scmDatanodeAddress;
+ this.blockClientAddress = Objects.requireNonNull(blockClientAddress,
"blockClientAddress == null");
+ this.scmClientAddress = Objects.requireNonNull(scmClientAddress,
"scmClientAddress == null");
+ this.scmSecurityAddress = Objects.requireNonNull(scmSecurityAddress,
"scmSecurityAddress == null");
+ this.scmDatanodeAddress = Objects.requireNonNull(scmDatanodeAddress,
"scmDatanodeAddress == null");
}
public String getServiceId() {
@@ -228,18 +229,22 @@ public String getNodeId() {
}
public String getBlockClientAddress() {
- return blockClientAddress;
+ return blockClientAddress.getHostAndPortString();
}
public String getScmClientAddress() {
- return scmClientAddress;
+ return scmClientAddress.getHostAndPortString();
}
public String getScmSecurityAddress() {
- return scmSecurityAddress;
+ return scmSecurityAddress.getHostAndPortString();
}
- public String getScmDatanodeAddress() {
+ public HostAndPort getScmDatanodeHostPortAddress() {
return scmDatanodeAddress;
}
+
+ public String getScmDatanodeAddress() {
+ return scmDatanodeAddress.getHostAndPortString();
+ }
}
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java
new file mode 100644
index 00000000000..e89ad73d31b
--- /dev/null
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java
@@ -0,0 +1,85 @@
+/*
+ * 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.hdds.scm.net;
+
+import java.net.InetSocketAddress;
+import org.apache.hadoop.net.NetUtils;
+
+/**
+ * A class for host and port.
+ * It also has an address which can be updated from time to time.
+ */
+public class HostAndPort {
+ private final String host;
+ private final int port;
+ private final String hostAndPortString;
+ private final int hash;
+ /** The address can be updated from time to time. */
+ private final InetSocketAddress address;
+
+ public HostAndPort(String host, int port) {
+ this.host = host;
+ this.port = port;
+ this.hostAndPortString = host + ":" + port;
+ this.hash = host.hashCode() ^ Integer.hashCode(port);
+ // TODO: HDDS-15533 change the address resolution logic and make
this.address threadsafe.
+ this.address = NetUtils.createSocketAddr(hostAndPortString);
+ }
+
+ public String getHostName() {
+ return host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public String getHostAndPortString() {
+ return hostAndPortString;
+ }
+
+ public InetSocketAddress getAddress() {
+ return address;
+ }
+
+ @Override
+ public int hashCode() {
+ return hash;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ } else if (!(obj instanceof HostAndPort)) {
+ return false;
+ }
+ final HostAndPort that = (HostAndPort) obj;
+ // address must not be compared
+ return this.hash == that.hash
+ && this.port == that.port
+ && this.host.equals(that.host);
+ }
+
+ @Override
+ public String toString() {
+ final InetSocketAddress a = getAddress();
+ final Object resolved = a != null && a.getAddress() != null ?
a.getAddress() : "<unresolved>";
+ return hostAndPortString + "/" + resolved;
+ }
+}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
index 00f1c055571..c2a30d97d52 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
@@ -39,7 +39,6 @@
import com.google.common.collect.Sets;
import java.io.File;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
@@ -71,6 +70,7 @@
import org.apache.hadoop.hdds.protocol.DiskBalancerProtocol;
import org.apache.hadoop.hdds.protocol.SecretKeyProtocol;
import
org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolClientSideTranslatorPB;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.security.SecurityConfig;
import org.apache.hadoop.hdds.security.symmetric.DefaultSecretKeyClient;
import org.apache.hadoop.hdds.security.symmetric.SecretKeyClient;
@@ -764,12 +764,12 @@ private String reconfigScmNodes(String value) {
LOG.info("Reconfiguring SCM nodes for service ID {} with new SCM nodes {}
and remove SCM nodes {}",
scmServiceId, scmNodesIdsToAdd, scmNodesIdsToRemove);
- Collection<Pair<String, InetSocketAddress>> scmToAdd =
HddsServerUtil.getSCMAddressForDatanodes(
+ final Collection<Pair<String, HostAndPort>> scmToAdd =
HddsServerUtil.getSCMAddressForDatanodes(
getConf(), scmServiceId, scmNodesIdsToAdd);
if (scmToAdd == null) {
throw new IllegalStateException("Reconfiguration failed to get SCM
address to add due to wrong configuration");
}
- Collection<Pair<String, InetSocketAddress>> scmToRemove =
HddsServerUtil.getSCMAddressForDatanodes(
+ final Collection<Pair<String, HostAndPort>> scmToRemove =
HddsServerUtil.getSCMAddressForDatanodes(
getConf(), scmServiceId, scmNodesIdsToRemove);
if (scmToRemove == null) {
throw new IllegalArgumentException(
@@ -790,10 +790,10 @@ private String reconfigScmNodes(String value) {
}
// Add the new SCM servers
- for (Pair<String, InetSocketAddress> pair : scmToAdd) {
+ for (Pair<String, HostAndPort> pair : scmToAdd) {
String scmNodeId = pair.getLeft();
- InetSocketAddress scmAddress = pair.getRight();
- if (scmAddress.isUnresolved()) {
+ final HostAndPort scmAddress = pair.getRight();
+ if (scmAddress.getAddress().isUnresolved()) {
LOG.warn("Reconfiguration failed to add SCM address {} for SCM service
{} since it can't " +
"be resolved, skipping", scmAddress, scmServiceId);
continue;
@@ -809,9 +809,9 @@ private String reconfigScmNodes(String value) {
}
// Remove the old SCM server
- for (Pair<String, InetSocketAddress> pair : scmToRemove) {
+ for (Pair<String, HostAndPort> pair : scmToRemove) {
String scmNodeId = pair.getLeft();
- InetSocketAddress scmAddress = pair.getRight();
+ final HostAndPort scmAddress = pair.getRight();
try {
connectionManager.removeSCMServer(scmAddress);
context.removeEndpoint(scmAddress);
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java
index d442b95285d..f376d640a3c 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java
@@ -21,11 +21,11 @@
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.CaseFormat;
-import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.text.WordUtils;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdfs.util.EnumCounters;
import org.apache.hadoop.metrics2.MetricsCollector;
import org.apache.hadoop.metrics2.MetricsInfo;
@@ -68,9 +68,9 @@ public final class DatanodeQueueMetrics implements
MetricsSource {
private Map<SCMCommandProto.Type, MetricsInfo> stateContextCommandQueueMap;
private Map<SCMCommandProto.Type, MetricsInfo> commandDispatcherQueueMap;
- private Map<InetSocketAddress, MetricsInfo> incrementalReportsQueueMap;
- private Map<InetSocketAddress, MetricsInfo> containerActionQueueMap;
- private Map<InetSocketAddress, MetricsInfo> pipelineActionQueueMap;
+ private Map<HostAndPort, MetricsInfo> incrementalReportsQueueMap;
+ private Map<HostAndPort, MetricsInfo> containerActionQueueMap;
+ private Map<HostAndPort, MetricsInfo> pipelineActionQueueMap;
public DatanodeQueueMetrics(DatanodeStateMachine datanodeStateMachine) {
this.registry = new MetricsRegistry(METRICS_SOURCE_NAME);
@@ -132,19 +132,19 @@ public void getMetrics(MetricsCollector collector,
boolean b) {
tmpEnum.get(entry.getKey()));
}
- for (Map.Entry<InetSocketAddress, MetricsInfo> entry:
+ for (Map.Entry<HostAndPort, MetricsInfo> entry:
incrementalReportsQueueMap.entrySet()) {
builder.addGauge(entry.getValue(),
datanodeStateMachine.getContext()
.getIncrementalReportQueueSize().getOrDefault(entry.getKey(),
0));
}
- for (Map.Entry<InetSocketAddress, MetricsInfo> entry:
+ for (Map.Entry<HostAndPort, MetricsInfo> entry:
containerActionQueueMap.entrySet()) {
builder.addGauge(entry.getValue(),
datanodeStateMachine.getContext()
.getContainerActionQueueSize().getOrDefault(entry.getKey(), 0));
}
- for (Map.Entry<InetSocketAddress, MetricsInfo> entry:
+ for (Map.Entry<HostAndPort, MetricsInfo> entry:
pipelineActionQueueMap.entrySet()) {
builder.addGauge(entry.getValue(),
datanodeStateMachine.getContext().getPipelineActionQueueSize()
@@ -157,7 +157,7 @@ public static synchronized void unRegister() {
DefaultMetricsSystem.instance().unregisterSource(METRICS_SOURCE_NAME);
}
- public void addEndpoint(InetSocketAddress endpoint) {
+ public void addEndpoint(HostAndPort endpoint) {
incrementalReportsQueueMap.computeIfAbsent(endpoint,
k -> getMetricsInfo(INCREMENTAL_REPORT_QUEUE_PREFIX,
CaseFormat.UPPER_UNDERSCORE
@@ -172,7 +172,7 @@ public void addEndpoint(InetSocketAddress endpoint) {
.to(CaseFormat.UPPER_CAMEL, k.getHostName())));
}
- public void removeEndpoint(InetSocketAddress endpoint) {
+ public void removeEndpoint(HostAndPort endpoint) {
incrementalReportsQueueMap.remove(endpoint);
containerActionQueueMap.remove(endpoint);
pipelineActionQueueMap.remove(endpoint);
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
index 94bc0549e66..b4a9a6e1a26 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
@@ -22,7 +22,6 @@
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.Closeable;
-import java.net.InetSocketAddress;
import java.time.ZonedDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -31,6 +30,7 @@
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.ozone.protocol.VersionResponse;
import org.apache.hadoop.ozone.protocolPB.ReconDatanodeProtocolPB;
import
org.apache.hadoop.ozone.protocolPB.StorageContainerDatanodeProtocolClientSideTranslatorPB;
@@ -46,7 +46,7 @@ public class EndpointStateMachine
LOG = LoggerFactory.getLogger(EndpointStateMachine.class);
private final StorageContainerDatanodeProtocolClientSideTranslatorPB
endPoint;
private final AtomicLong missedCount;
- private final InetSocketAddress address;
+ private final HostAndPort hostAndPort;
private final Lock lock;
private final ConfigurationSource conf;
private EndPointStates state = EndPointStates.FIRST;
@@ -64,18 +64,17 @@ public class EndpointStateMachine
*
* @param endPoint - RPC endPoint.
*/
- public EndpointStateMachine(InetSocketAddress address,
+ public EndpointStateMachine(HostAndPort hostAndPort,
StorageContainerDatanodeProtocolClientSideTranslatorPB endPoint,
ConfigurationSource conf, String threadNamePrefix) {
this.endPoint = endPoint;
this.missedCount = new AtomicLong(0);
- this.address = address;
+ this.hostAndPort = hostAndPort;
lock = new ReentrantLock();
this.conf = conf;
executorService = Executors.newSingleThreadExecutor(
new ThreadFactoryBuilder()
- .setNameFormat(threadNamePrefix + "EndpointStateMachineTaskThread-"
- + this.address + "-%d ")
+ .setNameFormat(threadNamePrefix +
"EndpointStateMachineTaskThread-" + hostAndPort + "-%d ")
.build());
}
@@ -180,7 +179,7 @@ public long getMissedCount() {
@Override
public String getAddressString() {
- return getAddress().toString();
+ return hostAndPort.getAddress().toString();
}
public void zeroMissedCount() {
@@ -192,8 +191,8 @@ public void zeroMissedCount() {
*
* @return - EndPoint.
*/
- public InetSocketAddress getAddress() {
- return this.address;
+ public HostAndPort getAddress() {
+ return hostAndPort;
}
/**
@@ -213,7 +212,7 @@ public InetSocketAddress getAddress() {
*/
@Override
public String toString() {
- return address.toString();
+ return hostAndPort.toString();
}
/**
@@ -236,14 +235,8 @@ public void logIfNeeded(Exception ex) {
long missedDurationSeconds = TimeUnit.MILLISECONDS.toSeconds(
this.getMissedCount() * getScmHeartbeatInterval(this.conf)
);
- LOG.warn(
- "Unable to communicate to {} server at {}:{} for past {}
seconds.",
- serverName,
- address.getAddress(),
- address.getPort(),
- missedDurationSeconds,
- ex
- );
+ LOG.warn("Unable to communicate to {} server at {} past {} seconds.",
+ serverName, hostAndPort, missedDurationSeconds, ex);
}
if (LOG.isTraceEnabled()) {
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java
index c7273dca741..25be82cd041 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java
@@ -24,7 +24,6 @@
import java.io.Closeable;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -36,6 +35,7 @@
import javax.management.ObjectName;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.retry.RetryPolicy;
@@ -61,7 +61,7 @@ public class SCMConnectionManager
LoggerFactory.getLogger(SCMConnectionManager.class);
private final ReadWriteLock mapLock;
- private final Map<InetSocketAddress, EndpointStateMachine> scmMachines;
+ private final Map<HostAndPort, EndpointStateMachine> scmMachines;
private final int rpcTimeout;
private final ConfigurationSource conf;
@@ -130,7 +130,7 @@ public void writeUnlock() {
* @param address - Address of the SCM machine to send heartbeat to.
* @throws IOException
*/
- public void addSCMServer(InetSocketAddress address,
+ public void addSCMServer(HostAndPort address,
String threadNamePrefix) throws IOException {
writeLock();
try {
@@ -156,7 +156,7 @@ public void addSCMServer(InetSocketAddress address,
StorageContainerDatanodeProtocolPB rpcProxy = RPC.getProtocolProxy(
StorageContainerDatanodeProtocolPB.class, version,
- address, UserGroupInformation.getCurrentUser(), hadoopConfig,
+ address.getAddress(), UserGroupInformation.getCurrentUser(),
hadoopConfig,
NetUtils.getDefaultSocketFactory(hadoopConfig), getRpcTimeout(),
retryPolicy).getProxy();
@@ -179,7 +179,7 @@ public void addSCMServer(InetSocketAddress address,
* @param address Recon address.
* @throws IOException
*/
- public void addReconServer(InetSocketAddress address,
+ public void addReconServer(HostAndPort address,
String threadNamePrefix) throws IOException {
LOG.info("Adding Recon Server : {}", address.toString());
writeLock();
@@ -202,7 +202,7 @@ public void addReconServer(InetSocketAddress address,
TimeUnit.MILLISECONDS);
ReconDatanodeProtocolPB rpcProxy = RPC.getProtocolProxy(
ReconDatanodeProtocolPB.class, version,
- address, UserGroupInformation.getCurrentUser(), hadoopConfig,
+ address.getAddress(), UserGroupInformation.getCurrentUser(),
hadoopConfig,
NetUtils.getDefaultSocketFactory(hadoopConfig), getRpcTimeout(),
retryPolicy).getProxy();
@@ -224,7 +224,7 @@ public void addReconServer(InetSocketAddress address,
* @param address - Address of the SCM machine to send heartbeat to.
* @throws IOException
*/
- public void removeSCMServer(InetSocketAddress address) throws IOException {
+ public void removeSCMServer(HostAndPort address) throws IOException {
writeLock();
try {
EndpointStateMachine endPoint = scmMachines.remove(address);
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
index 150159eb84a..4dcb74b40d5 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
@@ -28,7 +28,6 @@
import com.google.protobuf.Descriptors.Descriptor;
import com.google.protobuf.Message;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -70,6 +69,7 @@
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReport;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdfs.util.EnumCounters;
import
org.apache.hadoop.ozone.container.common.statemachine.commandhandler.ClosePipelineCommandHandler;
import org.apache.hadoop.ozone.container.common.states.DatanodeState;
@@ -112,16 +112,15 @@ public class StateContext {
private final DatanodeStateMachine parentDatanodeStateMachine;
private final AtomicLong stateExecutionCount;
private final ConfigurationSource conf;
- private final Set<InetSocketAddress> endpoints;
+ private final Set<HostAndPort> endpoints;
// Only the latest full report of each type is kept
private final AtomicReference<Message> containerReports;
private final AtomicReference<Message> nodeReport;
private final AtomicReference<Message> pipelineReports;
// Incremental reports are queued in the map below
- private final Map<InetSocketAddress, List<Message>>
- incrementalReportsQueue;
- private final Map<InetSocketAddress, Queue<ContainerAction>>
containerActions;
- private final Map<InetSocketAddress, PipelineActionMap> pipelineActions;
+ private final Map<HostAndPort, List<Message>> incrementalReportsQueue;
+ private final Map<HostAndPort, Queue<ContainerAction>> containerActions;
+ private final Map<HostAndPort, PipelineActionMap> pipelineActions;
private DatanodeStateMachine.DatanodeStates state;
private boolean shutdownOnError = false;
private boolean shutdownGracefully = false;
@@ -129,7 +128,7 @@ public class StateContext {
private final AtomicLong lastHeartbeatSent;
// Endpoint -> ReportType -> Boolean of whether the full report should be
// queued in getFullReports call.
- private final Map<InetSocketAddress,
+ private final Map<HostAndPort,
Map<String, AtomicBoolean>> isFullReportReadyToBeSent;
// List of supported full report types.
private final List<String> fullReportTypeList;
@@ -310,7 +309,7 @@ public void addIncrementalReport(Message report) {
// as an incremental message.
// see XceiverServerRatis#sendPipelineReport
synchronized (incrementalReportsQueue) {
- for (InetSocketAddress endpoint : endpoints) {
+ for (HostAndPort endpoint : endpoints) {
incrementalReportsQueue.get(endpoint).add(report);
}
}
@@ -349,7 +348,7 @@ public void refreshFullReport(Message report) {
* heartbeat.
*/
public void putBackReports(List<Message> reportsToPutBack,
- InetSocketAddress endpoint) {
+ HostAndPort endpoint) {
if (LOG.isDebugEnabled()) {
LOG.debug("endpoint: {}, size of reportsToPutBack: {}",
endpoint, reportsToPutBack.size());
@@ -375,8 +374,7 @@ public void putBackReports(List<Message> reportsToPutBack,
* @return List of reports
*/
public List<Message> getAllAvailableReports(
- InetSocketAddress endpoint
- ) {
+ HostAndPort endpoint) {
int maxLimit = Integer.MAX_VALUE;
// TODO: It is highly unlikely that we will reach maxLimit for the number
// for the number of reports, specially as it does not apply to the
@@ -400,7 +398,7 @@ public ContainerReportsProto
getFullContainerReportDiscardPendingICR()
synchronized (parentDatanodeStateMachine
.getContainer()) {
synchronized (incrementalReportsQueue) {
- for (Map.Entry<InetSocketAddress, List<Message>>
+ for (Map.Entry<HostAndPort, List<Message>>
entry : incrementalReportsQueue.entrySet()) {
if (entry.getValue() != null) {
entry.getValue().removeIf(
@@ -419,7 +417,7 @@ public ContainerReportsProto
getFullContainerReportDiscardPendingICR()
@VisibleForTesting
List<Message> getAllAvailableReportsUpToLimit(
- InetSocketAddress endpoint,
+ HostAndPort endpoint,
int limit) {
List<Message> reports = getFullReports(endpoint, limit);
List<Message> incrementalReports = getIncrementalReports(endpoint,
@@ -429,7 +427,7 @@ List<Message> getAllAvailableReportsUpToLimit(
}
List<Message> getIncrementalReports(
- InetSocketAddress endpoint, int maxLimit) {
+ HostAndPort endpoint, int maxLimit) {
List<Message> reportsToReturn = new LinkedList<>();
synchronized (incrementalReportsQueue) {
List<Message> reportsForEndpoint =
@@ -445,7 +443,7 @@ List<Message> getIncrementalReports(
}
List<Message> getFullReports(
- InetSocketAddress endpoint, int maxLimit) {
+ HostAndPort endpoint, int maxLimit) {
int count = 0;
Map<String, AtomicBoolean> mp = isFullReportReadyToBeSent.get(endpoint);
List<Message> fullReports = new LinkedList<>();
@@ -482,7 +480,7 @@ List<Message> getFullReports(
*/
public void addContainerAction(ContainerAction containerAction) {
synchronized (containerActions) {
- for (InetSocketAddress endpoint : endpoints) {
+ for (HostAndPort endpoint : endpoints) {
containerActions.get(endpoint).add(containerAction);
}
}
@@ -495,7 +493,7 @@ public void addContainerAction(ContainerAction
containerAction) {
*/
public void addContainerActionIfAbsent(ContainerAction containerAction) {
synchronized (containerActions) {
- for (InetSocketAddress endpoint : endpoints) {
+ for (HostAndPort endpoint : endpoints) {
if (!containerActions.get(endpoint).contains(containerAction)) {
containerActions.get(endpoint).add(containerAction);
}
@@ -510,7 +508,7 @@ public void addContainerActionIfAbsent(ContainerAction
containerAction) {
* @return {@literal List<ContainerAction>}
*/
public List<ContainerAction> getPendingContainerAction(
- InetSocketAddress endpoint,
+ HostAndPort endpoint,
int maxLimit) {
List<ContainerAction> containerActionList = new ArrayList<>();
synchronized (containerActions) {
@@ -538,7 +536,7 @@ public boolean addPipelineActionIfAbsent(PipelineAction
pipelineAction) {
// Put only if the pipeline id with the same action is absent.
final PipelineKey key = new PipelineKey(pipelineAction);
boolean added = false;
- for (InetSocketAddress endpoint : endpoints) {
+ for (HostAndPort endpoint : endpoints) {
added = pipelineActions.get(endpoint).putIfAbsent(key, pipelineAction)
|| added;
}
return added;
@@ -551,7 +549,7 @@ public boolean addPipelineActionIfAbsent(PipelineAction
pipelineAction) {
* @return {@literal List<ContainerAction>}
*/
public List<PipelineAction> getPendingPipelineAction(
- InetSocketAddress endpoint,
+ HostAndPort endpoint,
int maxLimit) {
final PipelineActionMap map = pipelineActions.get(endpoint);
if (map == null) {
@@ -894,7 +892,7 @@ public long getHeartbeatFrequency() {
return heartbeatFrequency.get();
}
- public void addEndpoint(InetSocketAddress endpoint) {
+ public void addEndpoint(HostAndPort endpoint) {
if (!endpoints.contains(endpoint)) {
this.endpoints.add(endpoint);
this.containerActions.put(endpoint, new LinkedList<>());
@@ -911,7 +909,7 @@ public void addEndpoint(InetSocketAddress endpoint) {
}
}
- public void removeEndpoint(InetSocketAddress endpoint) {
+ public void removeEndpoint(HostAndPort endpoint) {
this.endpoints.remove(endpoint);
this.containerActions.remove(endpoint);
this.pipelineActions.remove(endpoint);
@@ -948,17 +946,17 @@ public long getReconHeartbeatFrequency() {
return reconHeartbeatFrequency.get();
}
- public Map<InetSocketAddress, Integer> getPipelineActionQueueSize() {
+ public Map<HostAndPort, Integer> getPipelineActionQueueSize() {
return pipelineActions.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, e ->
e.getValue().size()));
}
- public Map<InetSocketAddress, Integer> getContainerActionQueueSize() {
+ public Map<HostAndPort, Integer> getContainerActionQueueSize() {
return containerActions.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, e ->
e.getValue().size()));
}
- public Map<InetSocketAddress, Integer> getIncrementalReportQueueSize() {
+ public Map<HostAndPort, Integer> getIncrementalReportQueueSize() {
return incrementalReportsQueue.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, e ->
e.getValue().size()));
}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java
index 2787093b1bf..d0c0a60b03e 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java
@@ -18,12 +18,10 @@
package org.apache.hadoop.ozone.container.common.states.datanode;
import static
org.apache.hadoop.hdds.utils.HddsServerUtil.getReconAddressForDatanodes;
-import static
org.apache.hadoop.hdds.utils.HddsServerUtil.getSCMAddressForDatanodes;
import com.google.common.base.Strings;
import java.io.File;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
@@ -33,6 +31,7 @@
import java.util.concurrent.TimeoutException;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.utils.HddsServerUtil;
import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
import
org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine;
@@ -76,9 +75,9 @@ public InitDatanodeState(ConfigurationSource conf,
*/
@Override
public DatanodeStateMachine.DatanodeStates call() throws Exception {
- Collection<InetSocketAddress> addresses = null;
+ final Collection<HostAndPort> addresses;
try {
- addresses = getSCMAddressForDatanodes(conf);
+ addresses = HddsServerUtil.getSCMAddressForDatanodes(conf);
} catch (IllegalArgumentException e) {
if (!Strings.isNullOrEmpty(e.getMessage())) {
LOG.error("Failed to get SCM addresses: {}", e.getMessage());
@@ -90,8 +89,8 @@ public DatanodeStateMachine.DatanodeStates call() throws
Exception {
LOG.error("Null or empty SCM address list found.");
return DatanodeStateMachine.DatanodeStates.SHUTDOWN;
} else {
- for (InetSocketAddress addr : addresses) {
- if (addr.isUnresolved()) {
+ for (HostAndPort addr : addresses) {
+ if (addr.getAddress().isUnresolved()) {
LOG.warn("One SCM address ({}) can't (yet?) be resolved. Postpone "
+ "initialization.", addr);
@@ -100,11 +99,11 @@ public DatanodeStateMachine.DatanodeStates call() throws
Exception {
return this.context.getState();
}
}
- for (InetSocketAddress addr : addresses) {
+ for (HostAndPort addr : addresses) {
connectionManager.addSCMServer(addr, context.getThreadNamePrefix());
this.context.addEndpoint(addr);
}
- InetSocketAddress reconAddress = getReconAddressForDatanodes(conf);
+ final HostAndPort reconAddress = getReconAddressForDatanodes(conf);
if (reconAddress != null) {
connectionManager.addReconServer(reconAddress,
context.getThreadNamePrefix());
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java
index fa3c209fa87..8231b2b3c7f 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java
@@ -46,6 +46,7 @@
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerType;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.security.token.TokenVerifier;
import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
import org.apache.hadoop.hdfs.util.Canceler;
@@ -150,7 +151,7 @@ public static EndpointStateMachine
createEndpoint(Configuration conf,
StorageContainerDatanodeProtocolClientSideTranslatorPB rpcClient =
new StorageContainerDatanodeProtocolClientSideTranslatorPB(rpcProxy);
- return new EndpointStateMachine(address, rpcClient,
+ return new EndpointStateMachine(new HostAndPort(address.getHostName(),
address.getPort()), rpcClient,
new LegacyHadoopConfigurationSource(conf), "");
}
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestSCMConnectionManager.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestSCMConnectionManager.java
index 12e62ffe733..0225b6aa873 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestSCMConnectionManager.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestSCMConnectionManager.java
@@ -19,8 +19,8 @@
import static
org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine.EndPointStates.HEARTBEAT;
-import java.net.InetSocketAddress;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -34,7 +34,7 @@ public void testRemoveSCMServerDoesNotMarkEndpointShutdown()
throws Exception {
try (SCMConnectionManager connectionManager =
new SCMConnectionManager(new OzoneConfiguration())) {
- InetSocketAddress address = new InetSocketAddress("127.0.0.1", 9861);
+ final HostAndPort address = new HostAndPort("127.0.0.1", 9861);
connectionManager.addSCMServer(address, "");
EndpointStateMachine endpoint =
connectionManager.getValues().iterator().next();
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java
index 8d79335591b..ca80b8b7751 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java
@@ -33,7 +33,6 @@
import com.google.protobuf.Descriptors.Descriptor;
import com.google.protobuf.Message;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -58,6 +57,7 @@
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReport;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdfs.util.EnumCounters;
import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
@@ -88,9 +88,9 @@ public void testPutBackReports() {
StateContext ctx = new StateContext(conf, DatanodeStates.getInitState(),
datanodeStateMachineMock, "");
- InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001);
+ HostAndPort scm1 = new HostAndPort("scm1", 9001);
ctx.addEndpoint(scm1);
- InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001);
+ HostAndPort scm2 = new HostAndPort("scm2", 9001);
ctx.addEndpoint(scm2);
Map<String, Integer> expectedReportCount = new HashMap<>();
@@ -142,9 +142,9 @@ public void testPutBackReports() {
@Test
public void testReportQueueWithAddReports() throws IOException {
StateContext ctx = createSubject();
- InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001);
+ HostAndPort scm1 = new HostAndPort("scm1", 9001);
ctx.addEndpoint(scm1);
- InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001);
+ HostAndPort scm2 = new HostAndPort("scm2", 9001);
ctx.addEndpoint(scm2);
// Check initial state
assertEquals(0, ctx.getAllAvailableReports(scm1).size());
@@ -303,9 +303,9 @@ private StateContext newStateContext(OzoneConfiguration
conf,
DatanodeStateMachine datanodeStateMachineMock) {
StateContext stateContext = new StateContext(conf,
DatanodeStates.getInitState(), datanodeStateMachineMock, "");
- InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001);
+ HostAndPort scm1 = new HostAndPort("scm1", 9001);
stateContext.addEndpoint(scm1);
- InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001);
+ HostAndPort scm2 = new HostAndPort("scm2", 9001);
stateContext.addEndpoint(scm2);
return stateContext;
}
@@ -332,8 +332,8 @@ public void testReportAPIs() {
StateContext stateContext = new StateContext(conf,
DatanodeStates.getInitState(), datanodeStateMachineMock, "");
- InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001);
- InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001);
+ HostAndPort scm1 = new HostAndPort("scm1", 9001);
+ HostAndPort scm2 = new HostAndPort("scm2", 9001);
Message generatedMessage =
newMockReport(StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME);
@@ -394,7 +394,7 @@ public void testClosePipelineActions() {
StateContext stateContext = new StateContext(conf,
DatanodeStates.getInitState(), datanodeStateMachineMock, "");
- InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001);
+ HostAndPort scm1 = new HostAndPort("scm1", 9001);
// Add SCM endpoint.
stateContext.addEndpoint(scm1);
@@ -452,8 +452,8 @@ public void testActionAPIs() {
StateContext stateContext = new StateContext(conf,
DatanodeStates.getInitState(), datanodeStateMachineMock, "");
- InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001);
- InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001);
+ HostAndPort scm1 = new HostAndPort("scm1", 9001);
+ HostAndPort scm2 = new HostAndPort("scm2", 9001);
// Try to get containerActions for endpoint which is not yet added.
List<ContainerAction> containerActions =
@@ -655,9 +655,9 @@ public void testGetReports() {
StateContext ctx = new StateContext(conf, DatanodeStates.getInitState(),
datanodeStateMachineMock, "");
- InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001);
+ HostAndPort scm1 = new HostAndPort("scm1", 9001);
ctx.addEndpoint(scm1);
- InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001);
+ HostAndPort scm2 = new HostAndPort("scm2", 9001);
ctx.addEndpoint(scm2);
// Check initial state
assertEquals(0, ctx.getAllAvailableReports(scm1).size());
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java
index e2b4fa167ea..12a38d3dcfb 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java
@@ -30,7 +30,6 @@
import static org.mockito.Mockito.when;
import com.google.protobuf.UnsafeByteOperations;
-import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -50,6 +49,7 @@
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager;
import org.apache.hadoop.hdfs.util.EnumCounters;
import
org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine;
@@ -67,8 +67,7 @@
*/
public class TestHeartbeatEndpointTask {
- private static final InetSocketAddress TEST_SCM_ENDPOINT =
- new InetSocketAddress("test-scm-1", 9861);
+ private static final HostAndPort TEST_SCM_ENDPOINT = new
HostAndPort("test-scm-1", 9861);
@Test
public void handlesReconstructContainerCommand() throws Exception {
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java
index e310eaf3dea..c2c6ec822ce 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java
@@ -32,7 +32,6 @@
import com.google.protobuf.Message;
import java.io.File;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
@@ -47,6 +46,7 @@
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.ozone.OzoneConfigKeys;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.ContainerTestHelper;
@@ -316,7 +316,7 @@ public void testVolumeFailure() throws IOException {
new OzoneConfiguration(), DatanodeStateMachine
.DatanodeStates.getInitState(),
datanodeStateMachineMock, "");
- InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001);
+ HostAndPort scm1 = new HostAndPort("scm1", 9001);
stateContext.addEndpoint(scm1);
when(datanodeStateMachineMock.getContainer()).thenReturn(ozoneContainer);
diff --git
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HddsServerUtil.java
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HddsServerUtil.java
index 760bdfbd04b..e9050346931 100644
---
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HddsServerUtil.java
+++
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HddsServerUtil.java
@@ -104,6 +104,7 @@
import org.apache.hadoop.hdds.recon.ReconConfigKeys;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol;
import org.apache.hadoop.hdds.scm.proxy.SCMClientConfig;
import
org.apache.hadoop.hdds.scm.proxy.SCMSecurityProtocolFailoverProxyProvider;
@@ -863,18 +864,14 @@ public static void setPoolSize(ThreadPoolExecutor
executor, int size, Logger log
* @return A collection of SCM addresses
* @throws IllegalArgumentException If the configuration is invalid
*/
- public static Collection<InetSocketAddress> getSCMAddressForDatanodes(
- ConfigurationSource conf) {
-
+ public static Collection<HostAndPort>
getSCMAddressForDatanodes(ConfigurationSource conf) {
// First check HA style config, if not defined fall back to OZONE_SCM_NAMES
if (getScmServiceId(conf) != null) {
List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
- Collection<InetSocketAddress> scmAddressList =
- new HashSet<>(scmNodeInfoList.size());
+ final Collection<HostAndPort> scmAddressList = new
HashSet<>(scmNodeInfoList.size());
for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
- scmAddressList.add(
- NetUtils.createSocketAddr(scmNodeInfo.getScmDatanodeAddress()));
+ scmAddressList.add(scmNodeInfo.getScmDatanodeHostPortAddress());
}
return scmAddressList;
} else {
@@ -887,7 +884,7 @@ public static Collection<InetSocketAddress>
getSCMAddressForDatanodes(
+ " Empty address list found.");
}
- Collection<InetSocketAddress> addresses = new HashSet<>(names.size());
+ final Collection<HostAndPort> addresses = new HashSet<>(names.size());
for (String address : names) {
Optional<String> hostname = getHostName(address);
if (!hostname.isPresent()) {
@@ -897,9 +894,7 @@ public static Collection<InetSocketAddress>
getSCMAddressForDatanodes(
int port = getHostPort(address)
.orElse(conf.getInt(OZONE_SCM_DATANODE_PORT_KEY,
OZONE_SCM_DATANODE_PORT_DEFAULT));
- InetSocketAddress addr = NetUtils.createSocketAddr(hostname.get(),
- port);
- addresses.add(addr);
+ addresses.add(new HostAndPort(hostname.get(), port));
}
if (addresses.size() > 1) {
@@ -920,9 +915,9 @@ public static Collection<InetSocketAddress>
getSCMAddressForDatanodes(
* Null if there is any wrongly configured SCM address. Note that the
returned collection
* might not be ordered the same way as the requested SCM node IDs
*/
- public static Collection<Pair<String, InetSocketAddress>>
getSCMAddressForDatanodes(
+ public static Collection<Pair<String, HostAndPort>>
getSCMAddressForDatanodes(
ConfigurationSource conf, String scmServiceId, Set<String> scmNodeIds) {
- Collection<Pair<String, InetSocketAddress>> scmNodeAddress = new
HashSet<>(scmNodeIds.size());
+ Collection<Pair<String, HostAndPort>> scmNodeAddress = new
HashSet<>(scmNodeIds.size());
for (String scmNodeId : scmNodeIds) {
String addressKey = ConfUtils.addKeySuffixes(
OZONE_SCM_ADDRESS_KEY, scmServiceId, scmNodeId);
@@ -936,9 +931,7 @@ public static Collection<Pair<String, InetSocketAddress>>
getSCMAddressForDatano
OZONE_SCM_DATANODE_ADDRESS_KEY, OZONE_SCM_DATANODE_PORT_KEY,
OZONE_SCM_DATANODE_PORT_DEFAULT);
- String scmDatanodeAddressStr = SCMNodeInfo.buildAddress(scmAddress,
scmDatanodePort);
- InetSocketAddress scmDatanodeAddress =
NetUtils.createSocketAddr(scmDatanodeAddressStr);
- scmNodeAddress.add(Pair.of(scmNodeId, scmDatanodeAddress));
+ scmNodeAddress.add(Pair.of(scmNodeId, new HostAndPort(scmAddress,
scmDatanodePort)));
}
return scmNodeAddress;
}
@@ -949,7 +942,7 @@ public static Collection<Pair<String, InetSocketAddress>>
getSCMAddressForDatano
* @return Recon address
* @throws IllegalArgumentException If the configuration is invalid
*/
- public static InetSocketAddress getReconAddressForDatanodes(
+ public static HostAndPort getReconAddressForDatanodes(
ConfigurationSource conf) {
String name = conf.get(OZONE_RECON_ADDRESS_KEY);
if (StringUtils.isEmpty(name)) {
@@ -961,6 +954,6 @@ public static InetSocketAddress getReconAddressForDatanodes(
+ name);
}
int port = getHostPort(name).orElse(OZONE_RECON_DATANODE_PORT_DEFAULT);
- return NetUtils.createSocketAddr(hostname.get(), port);
+ return new HostAndPort(hostname.get(), port);
}
}
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestHddsServerUtil.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestHddsServerUtil.java
index 2878304f386..46edbf13c8d 100644
---
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestHddsServerUtil.java
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestHddsServerUtil.java
@@ -36,8 +36,8 @@
import java.util.Map;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.utils.HddsServerUtil;
-import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.ozone.ha.ConfUtils;
import org.junit.jupiter.api.Test;
@@ -58,17 +58,15 @@ public void testGetScmDataNodeAddress() {
// First try a client address with just a host name. Verify it falls
// back to the default port.
conf.set(ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY, "1.2.3.4");
- InetSocketAddress addr = NetUtils.createSocketAddr(
- SCMNodeInfo.buildNodeInfo(conf).get(0).getScmDatanodeAddress());
- assertEquals("1.2.3.4", addr.getHostString());
+ HostAndPort addr =
SCMNodeInfo.buildNodeInfo(conf).get(0).getScmDatanodeHostPortAddress();
+ assertEquals("1.2.3.4", addr.getHostName());
assertEquals(ScmConfigKeys.OZONE_SCM_DATANODE_PORT_DEFAULT,
addr.getPort());
// Next try a client address with just a host name and port.
// Verify the port is ignored and the default DataNode port is used.
conf.set(ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY, "1.2.3.4:100");
- addr = NetUtils.createSocketAddr(
- SCMNodeInfo.buildNodeInfo(conf).get(0).getScmDatanodeAddress());
- assertEquals("1.2.3.4", addr.getHostString());
+ addr =
SCMNodeInfo.buildNodeInfo(conf).get(0).getScmDatanodeHostPortAddress();
+ assertEquals("1.2.3.4", addr.getHostName());
assertEquals(ScmConfigKeys.OZONE_SCM_DATANODE_PORT_DEFAULT,
addr.getPort());
// Set both OZONE_SCM_CLIENT_ADDRESS_KEY and
@@ -77,9 +75,8 @@ public void testGetScmDataNodeAddress() {
// default.
conf.set(ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY, "1.2.3.4:100");
conf.set(ScmConfigKeys.OZONE_SCM_DATANODE_ADDRESS_KEY, "5.6.7.8");
- addr = NetUtils.createSocketAddr(
- SCMNodeInfo.buildNodeInfo(conf).get(0).getScmDatanodeAddress());
- assertEquals("5.6.7.8", addr.getHostString());
+ addr =
SCMNodeInfo.buildNodeInfo(conf).get(0).getScmDatanodeHostPortAddress();
+ assertEquals("5.6.7.8", addr.getHostName());
assertEquals(ScmConfigKeys.OZONE_SCM_DATANODE_PORT_DEFAULT,
addr.getPort());
// Set both OZONE_SCM_CLIENT_ADDRESS_KEY and
@@ -88,9 +85,8 @@ public void testGetScmDataNodeAddress() {
// used.
conf.set(ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY, "1.2.3.4:100");
conf.set(ScmConfigKeys.OZONE_SCM_DATANODE_ADDRESS_KEY, "5.6.7.8:200");
- addr = NetUtils.createSocketAddr(
- SCMNodeInfo.buildNodeInfo(conf).get(0).getScmDatanodeAddress());
- assertEquals("5.6.7.8", addr.getHostString());
+ addr =
SCMNodeInfo.buildNodeInfo(conf).get(0).getScmDatanodeHostPortAddress();
+ assertEquals("5.6.7.8", addr.getHostName());
assertEquals(200, addr.getPort());
}
@@ -187,9 +183,9 @@ public void testScmDataNodeBindHostDefault() {
@Test
void testGetSCMAddresses() {
final OzoneConfiguration conf = new OzoneConfiguration();
- Collection<InetSocketAddress> addresses;
- InetSocketAddress addr;
- Iterator<InetSocketAddress> it;
+ Collection<HostAndPort> addresses;
+ HostAndPort addr;
+ Iterator<HostAndPort> it;
// Verify valid IP address setup
conf.setStrings(ScmConfigKeys.OZONE_SCM_NAMES, "1.2.3.4");
@@ -228,7 +224,7 @@ void testGetSCMAddresses() {
it = addresses.iterator();
HashMap<String, Integer> expected1 = new HashMap<>(hostsAndPorts);
while (it.hasNext()) {
- InetSocketAddress current = it.next();
+ HostAndPort current = it.next();
assertTrue(expected1.remove(current.getHostName(),
current.getPort()));
}
@@ -242,7 +238,7 @@ void testGetSCMAddresses() {
it = addresses.iterator();
HashMap<String, Integer> expected2 = new HashMap<>(hostsAndPorts);
while (it.hasNext()) {
- InetSocketAddress current = it.next();
+ HostAndPort current = it.next();
assertTrue(expected2.remove(current.getHostName(),
current.getPort()));
}
@@ -292,14 +288,14 @@ void testGetSCMAddressesWithHAConfig() {
expected.add("scm" + ":" + port);
}
- Collection<InetSocketAddress> scmAddressList =
+ Collection<HostAndPort> scmAddressList =
getSCMAddressForDatanodes(conf);
assertNotNull(scmAddressList);
assertEquals(3, scmAddressList.size());
- for (InetSocketAddress next : scmAddressList) {
- expected.remove(next.getHostName() + ":" + next.getPort());
+ for (HostAndPort next : scmAddressList) {
+ expected.remove(next.getHostAndPortString());
}
assertEquals(0, expected.size());
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestHddsServerUtils.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestHddsServerUtils.java
index d3142ebd96a..3d6e94e0a87 100644
---
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestHddsServerUtils.java
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestHddsServerUtils.java
@@ -54,6 +54,7 @@ public class TestHddsServerUtils {
public void testGetDatanodeAddressWithPort() {
final String scmHost = "host123:100";
final OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(OZONE_SCM_CLIENT_ADDRESS_KEY, scmHost);
conf.set(OZONE_SCM_DATANODE_ADDRESS_KEY, scmHost);
final InetSocketAddress address =
NetUtils.createSocketAddr(
@@ -69,6 +70,7 @@ public void testGetDatanodeAddressWithPort() {
public void testGetDatanodeAddressWithoutPort() {
final String scmHost = "host123";
final OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(OZONE_SCM_CLIENT_ADDRESS_KEY, scmHost);
conf.set(OZONE_SCM_DATANODE_ADDRESS_KEY, scmHost);
final InetSocketAddress address =
NetUtils.createSocketAddr(
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServer.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServer.java
index c60c63e3a7b..76858731d80 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServer.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServer.java
@@ -30,7 +30,6 @@
import com.google.inject.Guice;
import com.google.inject.Injector;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -40,6 +39,7 @@
import
org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolClientSideTranslatorPB;
import org.apache.hadoop.hdds.recon.ReconConfig;
import org.apache.hadoop.hdds.recon.ReconConfigKeys;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.scm.server.OzoneStorageContainerManager;
import org.apache.hadoop.hdds.security.SecurityConfig;
import
org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
@@ -397,7 +397,7 @@ private static void loginReconUser(OzoneConfiguration conf)
reconConfig.getKerberosPrincipal(),
reconConfig.getKerberosKeytab());
UserGroupInformation.setConfiguration(conf);
- InetSocketAddress socAddr =
HddsServerUtil.getReconAddressForDatanodes(conf);
+ final HostAndPort socAddr =
HddsServerUtil.getReconAddressForDatanodes(conf);
SecurityUtil.login(conf,
OZONE_RECON_KERBEROS_KEYTAB_FILE_KEY,
OZONE_RECON_KERBEROS_PRINCIPAL_KEY,
diff --git
a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/DatanodeSimulator.java
b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/DatanodeSimulator.java
index 324df519e59..3a42d57a01d 100644
---
a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/DatanodeSimulator.java
+++
b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/DatanodeSimulator.java
@@ -67,6 +67,7 @@
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMRegisteredResponseProto;
import
org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
import org.apache.hadoop.hdds.server.JsonUtils;
import org.apache.hadoop.hdds.server.ServerUtils;
@@ -425,13 +426,14 @@ private void heartbeat(InetSocketAddress endpoint,
private void init() throws IOException {
conf = freonCommand.getOzoneConf();
- Collection<InetSocketAddress> addresses = getSCMAddressForDatanodes(conf);
+ final Collection<HostAndPort> addresses = getSCMAddressForDatanodes(conf);
scmClients = new HashMap<>(addresses.size());
- for (InetSocketAddress address : addresses) {
+ for (HostAndPort a : addresses) {
+ InetSocketAddress address = a.getAddress();
scmClients.put(address, createScmClient(address));
}
- reconAddress = getReconAddressForDatanodes(conf);
+ reconAddress = getReconAddressForDatanodes(conf).getAddress();
reconClient = createReconClient(reconAddress);
heartbeatScheduler = Executors.newScheduledThreadPool(threadCount);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
