HDDS-15942. Remove Pair usage from hadoop-hdds module

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java

@@ -27,7 +27,6 @@
 import java.util.Arrays;
 import java.util.List;
 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;
@@ -382,10 +381,10 @@ private synchronized void readChunkFromContainer(int len) throws IOException {
     if (verifyChecksum) {
       // Adjust the chunk offset and length to include required checksum
       // boundaries
-      Pair<Long, Long> adjustedOffsetAndLength =
+      ChecksumBoundaries adjustedOffsetAndLength =
           computeChecksumBoundaries(startByteIndex, len);
-      adjustedBuffersOffset = adjustedOffsetAndLength.getLeft();
-      adjustedBuffersLen = adjustedOffsetAndLength.getRight();
+      adjustedBuffersOffset = adjustedOffsetAndLength.offset;
+      adjustedBuffersLen = adjustedOffsetAndLength.length;
     } else {
       // Read from the startByteIndex
       adjustedBuffersOffset = startByteIndex;
@@ -517,7 +516,7 @@ private void validateChunk(
    * @return Adjusted (Chunk Offset, Chunk Length) which needs to be read
    * from Container
    */
-  private Pair<Long, Long> computeChecksumBoundaries(long startByteIndex,
+  private ChecksumBoundaries computeChecksumBoundaries(long startByteIndex,
       int dataLen) {
 
     int bytesPerChecksum = chunkInfo.getChecksumData().getBytesPerChecksum();
@@ -529,7 +528,20 @@ 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);
+  }
+
+  /**
+   * Checksum-aligned chunk boundaries for a read operation.
+   */
+  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;
+    }
   }
 
   /**

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,11 @@ private void clearInternalBuffers() {
 
   protected void loadDataBuffersFromStream()
       throws IOException, InterruptedException {
-    Queue<ImmutablePair<Integer, Future<Void>>> pendingReads
+    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 +582,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 +592,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 +623,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) {

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/SafeModeRuleStatus.java

@@ -0,0 +1,68 @@
+/*
+ * 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;
+
+import java.util.Objects;
+
+/**
+ * Status of SCM safe mode exit rule.
+ */
+public final class SafeModeRuleStatus {
+
+  private final boolean validated;
+  private final String statusText;
+
+  public SafeModeRuleStatus(boolean validated, String statusText) {
+    this.validated = validated;
+    this.statusText = statusText;
+  }
+
+  public boolean isValidated() {
+    return validated;
+  }
+
+  public String getStatusText() {
+    return statusText;
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (this == other) {
+      return true;
+    }
+    if (!(other instanceof SafeModeRuleStatus)) {
+      return false;
+    }
+    SafeModeRuleStatus that = (SafeModeRuleStatus) other;
+    return validated == that.validated
+        && Objects.equals(statusText, that.statusText);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(validated, statusText);
+  }
+
+  @Override
+  public String toString() {
+    return "SafeModeRuleStatus{"
+        + "validated=" + validated
+        + ", statusText='" + statusText + '\''
+        + '}';
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java

@@ -25,7 +25,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
@@ -38,7 +37,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;
@@ -52,19 +51,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);
@@ -90,13 +89,26 @@ 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();
       }
     }
     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;
+    }
+  }
 }

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,7 +332,7 @@ private class Connection extends Thread {
     private IOException closeException; // close reason
 
     private final Thread rpcRequestThread;
-    private final SynchronousQueue<Pair<Call, ResponseBuffer>> rpcRequestQueue =
+    private final SynchronousQueue<RpcRequest> rpcRequestQueue =
         new SynchronousQueue<>(true);
 
     private AtomicReference<Thread> connectingThread = new AtomicReference<>();
@@ -1048,15 +1047,15 @@ public void run() {
         while (!shouldCloseConnection.get()) {
           ResponseBuffer buf = null;
           try {
-            Pair<Call, ResponseBuffer> pair =
+            RpcRequest rpcRequest =
                 rpcRequestQueue.poll(maxIdleTime, TimeUnit.MILLISECONDS);
-            if (pair == null || shouldCloseConnection.get()) {
+            if (rpcRequest == null || shouldCloseConnection.get()) {
               continue;
             }
-            buf = pair.getRight();
+            buf = rpcRequest.buffer;
             synchronized (ipcStreams.out) {
               if (LOG.isDebugEnabled()) {
-                Call call = pair.getLeft();
+                Call call = rpcRequest.call;
                 LOG.debug(getName() + "{} sending #{} {}", getName(), call.id,
                     call.rpcRequest);
               }
@@ -1115,12 +1114,25 @@ 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;
         }
       }
     }
 
+    /**
+     * A queued RPC call and its response buffer.
+     */
+    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 +1565,6 @@ void setAddress(InetSocketAddress address) {
       this.address = address;
     }
 
-
     Class<?> getProtocol() {
       return protocol;
     }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java

@@ -55,7 +55,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;
@@ -84,6 +83,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;
@@ -790,12 +790,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 Collection<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 Collection<ScmNodeAddress> scmToRemove = HddsServerUtil.getSCMAddressForDatanodes(
         getConf(), scmServiceId, scmNodesIdsToRemove);
     if (scmToRemove == null) {
       throw new IllegalArgumentException(
@@ -816,9 +816,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 entry : scmToAdd) {
+      String scmNodeId = entry.getScmNodeId();
+      final HostAndPort scmAddress = entry.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);
@@ -835,9 +835,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 entry : scmToRemove) {
+      String scmNodeId = entry.getScmNodeId();
+      final HostAndPort scmAddress = entry.getHostAndPort();
       try {
         connectionManager.removeSCMServer(scmAddress);
         context.removeEndpoint(scmAddress);