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 153902ae046 HDDS-16149. Remove Pair usage from
client/common/framework/container-service (#11132)
153902ae046 is described below
commit 153902ae046db42dc03cc14e6739411edbdf5268
Author: Navink <[email protected]>
AuthorDate: Wed Sep 23 01:05:50 2026 +0530
HDDS-16149. Remove Pair usage from
client/common/framework/container-service (#11132)
---
.../hadoop/hdds/scm/storage/ChunkInputStream.java | 26 ++++++++----
.../io/ECBlockReconstructedStripeInputStream.java | 25 ++++++++----
.../apache/hadoop/hdds/tracing/TraceAllMethod.java | 28 +++++++++----
.../main/java/org/apache/hadoop/ipc_/Client.java | 33 +++++++++------
.../apache/hadoop/ozone/HddsDatanodeService.java | 19 +++++----
.../checksum/ContainerMerkleTreeTestUtils.java | 34 ++++++++++++++--
.../container/checksum/TestContainerDiff.java | 19 +++++----
.../apache/hadoop/hdds/utils/HddsServerUtil.java | 18 ++++-----
.../apache/hadoop/hdds/utils/ScmNodeAddress.java | 47 ++++++++++++++++++++++
9 files changed, 180 insertions(+), 69 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
index f78adc85e41..67f769f56e4 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
@@ -28,7 +28,6 @@
import java.util.List;
import java.util.Objects;
import java.util.function.Supplier;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.fs.ByteBufferReadable;
import org.apache.hadoop.fs.CanUnbuffer;
import org.apache.hadoop.fs.Seekable;
@@ -403,9 +402,9 @@ ChunkInfo getChunkInfo(long chunkRelativePosition, int
toRead) {
final long adjustedOffset;
final long adjustedLen;
if (verifyChecksum) {
- Pair<Long, Long> boundaries =
computeChecksumBoundaries(chunkRelativePosition, toRead);
- adjustedOffset = boundaries.getLeft();
- adjustedLen = boundaries.getRight();
+ ChecksumBoundaries boundaries =
computeChecksumBoundaries(chunkRelativePosition, toRead);
+ adjustedOffset = boundaries.offset;
+ adjustedLen = boundaries.length;
} else {
adjustedOffset = chunkRelativePosition;
adjustedLen = toRead;
@@ -608,8 +607,7 @@ private void validateChunk(
* @return Adjusted (Chunk Offset, Chunk Length) which needs to be read
* from Container
*/
- private Pair<Long, Long> computeChecksumBoundaries(long startByteIndex,
- int dataLen) {
+ private ChecksumBoundaries computeChecksumBoundaries(long startByteIndex,
int dataLen) {
int bytesPerChecksum = chunkInfo.getChecksumData().getBytesPerChecksum();
// index of the last byte to be read from chunk, inclusively.
@@ -620,7 +618,21 @@ private Pair<Long, Long> computeChecksumBoundaries(long
startByteIndex,
final long endIndex = ((endByteIndex / bytesPerChecksum) + 1)
* bytesPerChecksum; // exclusive
long adjustedChunkLen = Math.min(endIndex, length) - adjustedChunkOffset;
- return Pair.of(adjustedChunkOffset, adjustedChunkLen);
+ return new ChecksumBoundaries(adjustedChunkOffset, adjustedChunkLen);
+ }
+
+ /**
+ * Represents a byte range (offset and length) expanded to align with
+ * checksum chunk boundaries required for verification.
+ */
+ private static final class ChecksumBoundaries {
+ private final long offset;
+ private final long length;
+
+ private ChecksumBoundaries(long offset, long length) {
+ this.offset = offset;
+ this.length = length;
+ }
}
/**
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
index 672894f186f..5540c916b4a 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
@@ -43,7 +43,6 @@
import java.util.concurrent.Future;
import java.util.function.Function;
import org.apache.commons.lang3.NotImplementedException;
-import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -570,11 +569,10 @@ private void clearInternalBuffers() {
protected void loadDataBuffersFromStream()
throws IOException, InterruptedException {
- Queue<ImmutablePair<Integer, Future<Void>>> pendingReads
- = new ArrayDeque<>();
+ Queue<PendingRead> pendingReads = new ArrayDeque<>();
for (int i : selectedIndexes) {
ByteBuffer buf = decoderInputBuffers[i];
- pendingReads.add(new ImmutablePair<>(i, executor.submit(() -> {
+ pendingReads.add(new PendingRead(i, executor.submit(() -> {
readIntoBuffer(i, buf);
return null;
})));
@@ -583,8 +581,8 @@ protected void loadDataBuffersFromStream()
while (!pendingReads.isEmpty()) {
int index = -1;
try {
- ImmutablePair<Integer, Future<Void>> pair = pendingReads.poll();
- index = pair.getKey();
+ PendingRead pendingRead = pendingReads.poll();
+ index = pendingRead.index;
// Should this future.get() have a timeout? At the end of the call
chain
// we eventually call a grpc or ratis client to read the block data.
Its
// the call to the DNs which could potentially block. There is a
timeout
@@ -593,7 +591,7 @@ protected void loadDataBuffersFromStream()
// Which defaults to 30s. So if there is a DN communication problem, it
// should timeout in the client which should propagate up the stack as
// an IOException.
- pair.getValue().get();
+ pendingRead.future.get();
} catch (ExecutionException ee) {
boolean added = failedDataIndexes.add(index);
Throwable t = ee.getCause() != null ? ee.getCause() : ee;
@@ -624,6 +622,19 @@ protected void loadDataBuffersFromStream()
}
}
+ /**
+ * A pending block read tracked by stripe reconstruction.
+ */
+ private static final class PendingRead {
+ private final int index;
+ private final Future<Void> future;
+
+ private PendingRead(int index, Future<Void> future) {
+ this.index = index;
+ this.future = future;
+ }
+ }
+
private void readIntoBuffer(int ind, ByteBuffer buf) throws IOException {
List<DatanodeDetails> failedLocations = new LinkedList<>();
while (true) {
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java
index b2d4f8ade73..2acb46d11f8 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java
@@ -26,7 +26,6 @@
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
-import org.apache.commons.lang3.tuple.Pair;
/**
* A Java proxy invocation handler to trace all the methods of the delegate
@@ -39,7 +38,7 @@ public class TraceAllMethod<T> implements InvocationHandler {
/**
* Cache for all the method objects of the delegate class.
*/
- private final Map<String, Map<Class<?>[], Pair<Boolean, Method>>> methods =
new HashMap<>();
+ private final Map<String, Map<Class<?>[], DelegatedMethodInfo>> methods =
new HashMap<>();
private final T delegate;
private final String name;
@@ -53,19 +52,19 @@ public TraceAllMethod(T delegate, String name) {
}
boolean shouldSkip = method.isAnnotationPresent(SkipTracing.class);
methods.computeIfAbsent(method.getName(), any -> new HashMap<>())
- .put(method.getParameterTypes(), Pair.of(shouldSkip, method));
+ .put(method.getParameterTypes(), new DelegatedMethodInfo(shouldSkip,
method));
}
}
@Override
public Object invoke(Object proxy, Method method, Object[] args)
throws Throwable {
- Pair<Boolean, Method> methodInfo = findDelegatedMethod(method);
+ DelegatedMethodInfo methodInfo = findDelegatedMethod(method);
if (methodInfo == null) {
throw new NoSuchMethodException("Method not found: " + method.getName());
}
- boolean shouldSkip = methodInfo.getLeft();
- Method delegateMethod = methodInfo.getRight();
+ boolean shouldSkip = methodInfo.shouldSkip;
+ Method delegateMethod = methodInfo.delegateMethod;
if (shouldSkip) {
try {
return delegateMethod.invoke(delegate, args);
@@ -97,8 +96,8 @@ public Object invoke(Object proxy, Method method, Object[]
args)
}
}
- private Pair<Boolean, Method> findDelegatedMethod(Method method) {
- for (Entry<Class<?>[], Pair<Boolean, Method>> entry : methods.getOrDefault(
+ private DelegatedMethodInfo findDelegatedMethod(Method method) {
+ for (Entry<Class<?>[], DelegatedMethodInfo> entry : methods.getOrDefault(
method.getName(), emptyMap()).entrySet()) {
if (Arrays.equals(entry.getKey(), method.getParameterTypes())) {
return entry.getValue();
@@ -106,4 +105,17 @@ private Pair<Boolean, Method> findDelegatedMethod(Method
method) {
}
return null;
}
+
+ /**
+ * Whether a method should be skipped and the delegate method to invoke.
+ */
+ private static final class DelegatedMethodInfo {
+ private final boolean shouldSkip;
+ private final Method delegateMethod;
+
+ private DelegatedMethodInfo(boolean shouldSkip, Method delegateMethod) {
+ this.shouldSkip = shouldSkip;
+ this.delegateMethod = delegateMethod;
+ }
+ }
}
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/Client.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/Client.java
index e54844672b2..3497cec258e 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/Client.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/Client.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.ipc_;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.security.AccessControlException;
import com.google.common.base.Preconditions;
import org.apache.hadoop.conf.Configuration;
@@ -333,8 +332,7 @@ private class Connection extends Thread {
private IOException closeException; // close reason
private final Thread rpcRequestThread;
- private final SynchronousQueue<Pair<Call, ResponseBuffer>> rpcRequestQueue
=
- new SynchronousQueue<>(true);
+ private final SynchronousQueue<RpcRequest> rpcRequestQueue = new
SynchronousQueue<>(true);
private AtomicReference<Thread> connectingThread = new AtomicReference<>();
private final Consumer<Connection> removeMethod;
@@ -1048,17 +1046,15 @@ public void run() {
while (!shouldCloseConnection.get()) {
ResponseBuffer buf = null;
try {
- Pair<Call, ResponseBuffer> pair =
- rpcRequestQueue.poll(maxIdleTime, TimeUnit.MILLISECONDS);
- if (pair == null || shouldCloseConnection.get()) {
+ RpcRequest rpcRequest = rpcRequestQueue.poll(maxIdleTime,
TimeUnit.MILLISECONDS);
+ if (rpcRequest == null || shouldCloseConnection.get()) {
continue;
}
- buf = pair.getRight();
+ buf = rpcRequest.buffer;
synchronized (ipcStreams.out) {
if (LOG.isDebugEnabled()) {
- Call call = pair.getLeft();
- LOG.debug(getName() + "{} sending #{} {}", getName(), call.id,
- call.rpcRequest);
+ Call call = rpcRequest.call;
+ LOG.debug(getName() + "{} sending #{} {}", getName(), call.id,
call.rpcRequest);
}
// RpcRequestHeader + RpcRequest
ipcStreams.sendRequest(buf.toByteArray());
@@ -1115,12 +1111,26 @@ public void sendRpcRequest(final Call call)
// prevent a race condition between checking the shouldCloseConnection
// and the stopping of the polling thread
while (!shouldCloseConnection.get()) {
- if (rpcRequestQueue.offer(Pair.of(call, buf), 1, TimeUnit.SECONDS)) {
+ if (rpcRequestQueue.offer(new RpcRequest(call, buf), 1,
TimeUnit.SECONDS)) {
break;
}
}
}
+ /**
+ * Holds an active RPC call and its pre-serialized outbound payload
+ * ready for wire transmission.
+ */
+ private final class RpcRequest {
+ private final Call call;
+ private final ResponseBuffer buffer;
+
+ private RpcRequest(Call call, ResponseBuffer buffer) {
+ this.call = call;
+ this.buffer = buffer;
+ }
+ }
+
/* Receive a response.
* Because only one receiver, so no synchronization on in.
*/
@@ -1553,7 +1563,6 @@ void setAddress(InetSocketAddress address) {
this.address = address;
}
-
Class<?> getProtocol() {
return protocol;
}
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 e2023a15a35..42af80f8051 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
@@ -40,7 +40,6 @@
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
-import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -54,7 +53,6 @@
import java.util.stream.Stream;
import javax.management.ObjectName;
import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.hdds.DatanodeVersion;
import org.apache.hadoop.hdds.HddsConfigKeys;
@@ -83,6 +81,7 @@
import org.apache.hadoop.hdds.utils.HddsServerUtil;
import org.apache.hadoop.hdds.utils.HddsVersionInfo;
import org.apache.hadoop.hdds.utils.IOUtils;
+import org.apache.hadoop.hdds.utils.ScmNodeAddress;
import org.apache.hadoop.metrics2.util.MBeans;
import org.apache.hadoop.ozone.container.common.DatanodeLayoutStorage;
import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
@@ -783,12 +782,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);
- final Collection<Pair<String, HostAndPort>> scmToAdd =
HddsServerUtil.getSCMAddressForDatanodes(
+ final List<ScmNodeAddress> scmToAdd =
HddsServerUtil.getSCMAddressForDatanodes(
getConf(), scmServiceId, scmNodesIdsToAdd);
if (scmToAdd == null) {
throw new IllegalStateException("Reconfiguration failed to get SCM
address to add due to wrong configuration");
}
- final Collection<Pair<String, HostAndPort>> scmToRemove =
HddsServerUtil.getSCMAddressForDatanodes(
+ final List<ScmNodeAddress> scmToRemove =
HddsServerUtil.getSCMAddressForDatanodes(
getConf(), scmServiceId, scmNodesIdsToRemove);
if (scmToRemove == null) {
throw new IllegalArgumentException(
@@ -809,9 +808,9 @@ private String reconfigScmNodes(String value) {
}
// Add the new SCM servers
- for (Pair<String, HostAndPort> pair : scmToAdd) {
- String scmNodeId = pair.getLeft();
- final HostAndPort scmAddress = pair.getRight();
+ for (ScmNodeAddress nodeAddress : scmToAdd) {
+ String scmNodeId = nodeAddress.getScmNodeId();
+ final HostAndPort scmAddress = nodeAddress.getHostAndPort();
if (scmAddress.getAddress().isUnresolved()) {
LOG.warn("Reconfiguration failed to add SCM address {} for SCM service
{} since it can't " +
"be resolved, skipping", scmAddress, scmServiceId);
@@ -828,9 +827,9 @@ private String reconfigScmNodes(String value) {
}
// Remove the old SCM server
- for (Pair<String, HostAndPort> pair : scmToRemove) {
- String scmNodeId = pair.getLeft();
- final HostAndPort scmAddress = pair.getRight();
+ for (ScmNodeAddress nodeAddress : scmToRemove) {
+ String scmNodeId = nodeAddress.getScmNodeId();
+ final HostAndPort scmAddress = nodeAddress.getHostAndPort();
try {
connectionManager.removeSCMServer(scmAddress);
context.removeEndpoint(scmAddress);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/ContainerMerkleTreeTestUtils.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/ContainerMerkleTreeTestUtils.java
index 444ab7eef28..f33d23afba2 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/ContainerMerkleTreeTestUtils.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/ContainerMerkleTreeTestUtils.java
@@ -36,7 +36,6 @@
import java.util.Map;
import java.util.Random;
import java.util.stream.Collectors;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -183,9 +182,9 @@ public static List<BlockData>
getDeletedBlockData(ConfigurationSource conf, long
}
/**
- * Returns a Pair of merkle tree and the expected container diff for that
merkle tree.
+ * Returns the merkle tree and expected diff for that tree.
*/
- public static Pair<ContainerProtos.ContainerMerkleTree, ContainerDiffReport>
+ static ContainerDiffResult
buildTestTreeWithMismatches(ContainerMerkleTreeWriter originalTree, int
numMissingBlocks, int numMissingChunks,
int numCorruptChunks) {
@@ -196,7 +195,7 @@ public static List<BlockData>
getDeletedBlockData(ConfigurationSource conf, long
introduceMissingChunks(treeBuilder, numMissingChunks, diff);
introduceCorruptChunks(treeBuilder, numCorruptChunks, diff);
ContainerProtos.ContainerMerkleTree build = treeBuilder.build();
- return Pair.of(build, diff);
+ return new ContainerDiffResult(build, diff);
}
/**
@@ -421,4 +420,31 @@ public static BlockData buildBlockData(ConfigurationSource
config, long containe
blockData.addChunk(buildChunk(config, 2, ByteBuffer.wrap(new
byte[]{byteValue++, byteValue++, byteValue++})));
return blockData;
}
+
+ /**
+ * Result of building a test merkle tree with mismatches.
+ */
+ static final class ContainerDiffResult {
+ private final ContainerProtos.ContainerMerkleTree tree;
+ private final ContainerDiffReport diff;
+
+ private ContainerDiffResult(ContainerProtos.ContainerMerkleTree tree,
+ ContainerDiffReport diff) {
+ this.tree = tree;
+ this.diff = diff;
+ }
+
+ public ContainerProtos.ContainerMerkleTree getTree() {
+ return tree;
+ }
+
+ public ContainerDiffReport getDiff() {
+ return diff;
+ }
+
+ @Override
+ public String toString() {
+ return "ContainerDiffResult{" + "tree=" + tree + ", diff=" + diff + '}';
+ }
+ }
}
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/TestContainerDiff.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/TestContainerDiff.java
index c1331ab65bb..e5a341338ac 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/TestContainerDiff.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/TestContainerDiff.java
@@ -33,7 +33,6 @@
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -106,10 +105,10 @@ public static Stream<Arguments>
getContainerDiffMismatches() {
public void testContainerDiffWithMismatches(int numMissingBlock, int
numMissingChunk,
int numCorruptChunk) throws
Exception {
ContainerMerkleTreeWriter peerMerkleTree = buildTestTree(config);
- Pair<ContainerProtos.ContainerMerkleTree, ContainerDiffReport> buildResult
=
+ ContainerMerkleTreeTestUtils.ContainerDiffResult buildResult =
buildTestTreeWithMismatches(peerMerkleTree, numMissingBlock,
numMissingChunk, numCorruptChunk);
- ContainerDiffReport expectedDiff = buildResult.getRight();
- ContainerProtos.ContainerMerkleTree ourMerkleTree = buildResult.getLeft();
+ ContainerDiffReport expectedDiff = buildResult.getDiff();
+ ContainerProtos.ContainerMerkleTree ourMerkleTree = buildResult.getTree();
updateTreeProto(container, ourMerkleTree);
ContainerProtos.ContainerChecksumInfo peerChecksumInfo =
ContainerProtos.ContainerChecksumInfo.newBuilder()
.setContainerID(container.getContainerID())
@@ -134,9 +133,9 @@ public void testContainerDiffWithMismatches(int
numMissingBlock, int numMissingC
public void testPeerWithMismatchesHasNoDiff(int numMissingBlock, int
numMissingChunk,
int numCorruptChunk) throws
Exception {
ContainerMerkleTreeWriter ourMerkleTree = buildTestTree(config);
- Pair<ContainerProtos.ContainerMerkleTree, ContainerDiffReport> buildResult
=
+ ContainerMerkleTreeTestUtils.ContainerDiffResult buildResult =
buildTestTreeWithMismatches(ourMerkleTree, numMissingBlock,
numMissingChunk, numCorruptChunk);
- ContainerProtos.ContainerMerkleTree peerMerkleTree =
buildResult.getLeft();
+ ContainerProtos.ContainerMerkleTree peerMerkleTree =
buildResult.getTree();
checksumManager.updateTree(container, ourMerkleTree);
ContainerProtos.ContainerChecksumInfo peerChecksumInfo =
ContainerProtos.ContainerChecksumInfo.newBuilder()
.setContainerID(container.getContainerID())
@@ -173,7 +172,7 @@ void testContainerDiffWithBlockDeletionInPeer() throws
Exception {
// Create only 5 blocks in our tree. The peer has 5 more blocks that it
has deleted.
ContainerMerkleTreeWriter dummy = buildTestTree(config, 5);
// Introduce block corruption in our merkle tree.
- ContainerProtos.ContainerMerkleTree ourMerkleTree =
buildTestTreeWithMismatches(dummy, 3, 3, 3).getLeft();
+ ContainerProtos.ContainerMerkleTree ourMerkleTree =
buildTestTreeWithMismatches(dummy, 3, 3, 3).getTree();
ContainerProtos.ContainerChecksumInfo.Builder peerChecksumInfoBuilder =
ContainerProtos.ContainerChecksumInfo
.newBuilder()
@@ -217,7 +216,7 @@ void testDeletedBlocksInPeerAndBoth() throws Exception {
ContainerProtos.ContainerMerkleTree peerMerkleTree = buildTestTree(config,
5, 1, 2, 3, 4, 5);
// Introduce missing blocks in our merkle tree
ContainerProtos.ContainerMerkleTree ourMerkleTree =
- buildTestTreeWithMismatches(new
ContainerMerkleTreeWriter(peerMerkleTree), 3, 0, 0).getLeft();
+ buildTestTreeWithMismatches(new
ContainerMerkleTreeWriter(peerMerkleTree), 3, 0, 0).getTree();
// List<ContainerProtos.BlockMerkleTree> deletedBlockList = new
ArrayList<>();
// List<Long> blockIDs = Arrays.asList(1L, 2L, 3L, 4L, 5L);
@@ -262,7 +261,7 @@ void testDeletedBlocksInOurContainerOnly() throws Exception
{
// Setup deleted blocks only in the peer container checksum
ContainerMerkleTreeWriter peerMerkleTree = buildTestTree(config);
// Introduce block corruption in our merkle tree.
- ContainerProtos.ContainerMerkleTree ourMerkleTree =
buildTestTreeWithMismatches(peerMerkleTree, 0, 3, 3).getLeft();
+ ContainerProtos.ContainerMerkleTree ourMerkleTree =
buildTestTreeWithMismatches(peerMerkleTree, 0, 3, 3).getTree();
ContainerProtos.ContainerChecksumInfo peerChecksumInfo =
ContainerProtos.ContainerChecksumInfo
.newBuilder().setContainerMerkleTree(peerMerkleTree.toProto()).setContainerID(CONTAINER_ID).build();
@@ -289,7 +288,7 @@ void testCorruptionInOurMerkleTreeAndDeletedBlocksInPeer()
throws Exception {
ContainerProtos.ContainerMerkleTree peerMerkleTree = buildTestTree(config,
5, 1, 2, 3, 4, 5);
// Create our tree the same as the peer, but introduce corruption instead
of deleting blocks.
ContainerProtos.ContainerMerkleTree ourMerkleTree =
- buildTestTreeWithMismatches(buildTestTree(config, 5), 0, 3,
3).getLeft();
+ buildTestTreeWithMismatches(buildTestTree(config, 5), 0, 3,
3).getTree();
ContainerProtos.ContainerChecksumInfo peerChecksumInfo =
ContainerProtos.ContainerChecksumInfo
.newBuilder()
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 49a71025c76..e2d5536ffaf 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
@@ -85,7 +85,6 @@
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.SystemUtils;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.HddsUtils;
@@ -435,7 +434,6 @@ public static long getStaleNodeInterval(ConfigurationSource
conf) {
long heartbeatIntervalMs = getScmHeartbeatInterval(conf);
-
// Make sure that StaleNodeInterval is configured way above the frequency
// at which we run the heartbeat thread.
//
@@ -922,16 +920,14 @@ public static Collection<HostAndPort>
getSCMAddressForDatanodes(ConfigurationSou
* @param conf Configuration
* @param scmServiceId SCM service ID
* @param scmNodeIds Requested SCM node IDs
- * @return A collection with addresses of the request SCM node IDs.
- * 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
+ * @return A list with addresses of the requested SCM node IDs, in the same
iteration
+ * order as scmNodeIds. Null if there is any wrongly configured SCM address.
*/
- public static Collection<Pair<String, HostAndPort>>
getSCMAddressForDatanodes(
+ public static List<ScmNodeAddress> getSCMAddressForDatanodes(
ConfigurationSource conf, String scmServiceId, Set<String> scmNodeIds) {
- Collection<Pair<String, HostAndPort>> scmNodeAddress = new
HashSet<>(scmNodeIds.size());
+ List<ScmNodeAddress> scmNodeAddresses = new ArrayList<>(scmNodeIds.size());
for (String scmNodeId : scmNodeIds) {
- String addressKey = ConfUtils.addKeySuffixes(
- OZONE_SCM_ADDRESS_KEY, scmServiceId, scmNodeId);
+ String addressKey = ConfUtils.addKeySuffixes(OZONE_SCM_ADDRESS_KEY,
scmServiceId, scmNodeId);
String scmAddress = conf.get(addressKey);
if (scmAddress == null) {
LOG.warn("The SCM address configuration {} is not defined, return
nothing", addressKey);
@@ -942,9 +938,9 @@ public static Collection<Pair<String, HostAndPort>>
getSCMAddressForDatanodes(
OZONE_SCM_DATANODE_ADDRESS_KEY, OZONE_SCM_DATANODE_PORT_KEY,
OZONE_SCM_DATANODE_PORT_DEFAULT);
- scmNodeAddress.add(Pair.of(scmNodeId, new HostAndPort(scmAddress,
scmDatanodePort)));
+ scmNodeAddresses.add(new ScmNodeAddress(scmNodeId, new
HostAndPort(scmAddress, scmDatanodePort)));
}
- return scmNodeAddress;
+ return scmNodeAddresses;
}
/**
diff --git
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/ScmNodeAddress.java
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/ScmNodeAddress.java
new file mode 100644
index 00000000000..7f29a322459
--- /dev/null
+++
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/ScmNodeAddress.java
@@ -0,0 +1,47 @@
+/*
+ * 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.utils;
+
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
+
+/**
+ * SCM node identifier and its datanode address.
+ */
+public final class ScmNodeAddress {
+
+ private final String scmNodeId;
+ private final HostAndPort hostAndPort;
+
+ public ScmNodeAddress(String scmNodeId, HostAndPort hostAndPort) {
+ this.scmNodeId = scmNodeId;
+ this.hostAndPort = hostAndPort;
+ }
+
+ public String getScmNodeId() {
+ return scmNodeId;
+ }
+
+ public HostAndPort getHostAndPort() {
+ return hostAndPort;
+ }
+
+ @Override
+ public String toString() {
+ return "ScmNodeAddress{" + "scmNodeId='" + scmNodeId + '\'' + ",
hostAndPort=" + hostAndPort + '}';
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
