Merge branch 'apache:master' into HDDS-11463

Author: slfan1989

.github/workflows/check.yml

@@ -167,7 +167,6 @@ jobs:
         with:
           path: |
             ~/.pnpm-store
-            **/node_modules
           key: ${{ runner.os }}-pnpm-${{ hashFiles('**/pnpm-lock.yaml') }}
           restore-keys: |
             ${{ runner.os }}-pnpm-

dev-support/byteman/container-state-template.btm

@@ -0,0 +1,24 @@
+# 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.
+
+RULE Change container state to STATE_PLACEHOLDER
+CLASS org.apache.hadoop.ozone.container.common.impl.ContainerData
+METHOD getState
+AT ENTRY
+IF TRUE
+DO
+  traceln("BYTEMAN RULE: Overriding getState() to return STATE_PLACEHOLDER state");
+  return org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos$ContainerDataProto$State.valueOf("STATE_PLACEHOLDER")
+ENDRULE

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

@@ -51,16 +51,19 @@ public class XceiverClientMetrics implements MetricsSource {
   private EnumMap<ContainerProtos.Type, MutableCounterLong> pendingOpsArray;
   private EnumMap<ContainerProtos.Type, MutableCounterLong> opsArray;
   private EnumMap<ContainerProtos.Type, PerformanceMetrics> containerOpsLatency;
+
+  // TODO: https://issues.apache.org/jira/browse/HDDS-13555
+  @SuppressWarnings("PMD.SingularField")
   private MetricsRegistry registry;
-  private OzoneConfiguration conf = new OzoneConfiguration();
-  private int[] intervals = conf.getInts(OzoneConfigKeys
-      .OZONE_XCEIVER_CLIENT_METRICS_PERCENTILES_INTERVALS_SECONDS_KEY);
 
   public XceiverClientMetrics() {
     init();
   }
 
   public void init() {
+    OzoneConfiguration conf = new OzoneConfiguration();
+    int[] intervals = conf.getInts(OzoneConfigKeys.OZONE_XCEIVER_CLIENT_METRICS_PERCENTILES_INTERVALS_SECONDS_KEY);
+
     this.registry = new MetricsRegistry(SOURCE_NAME);
 
     this.pendingOpsArray = new EnumMap<>(ContainerProtos.Type.class);

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

@@ -125,8 +125,6 @@ public class BlockDataStreamOutput implements ByteBufferStreamOutput {
   private final List<DatanodeDetails> failedServers;
   private final Checksum checksum;
 
-  //number of buffers used before doing a flush/putBlock.
-  private int flushPeriod;
   private final Token<? extends TokenIdentifier> token;
   private final String tokenString;
   private final DataStreamOutput out;
@@ -172,8 +170,9 @@ public BlockDataStreamOutput(
     // Alternatively, stream setup can be delayed till the first chunk write.
     this.out = setupStream(pipeline);
     this.bufferList = bufferList;
-    flushPeriod = (int) (config.getStreamBufferFlushSize() / config
-        .getStreamBufferSize());
+
+    //number of buffers used before doing a flush/putBlock.
+    int flushPeriod = (int) (config.getStreamBufferFlushSize() / config.getStreamBufferSize());
 
     Preconditions
         .checkArgument(

hadoop-hdds/common/pom.xml

@@ -55,10 +55,6 @@
       <groupId>com.google.protobuf</groupId>
       <artifactId>protobuf-java</artifactId>
     </dependency>
-    <dependency>
-      <groupId>commons-collections</groupId>
-      <artifactId>commons-collections</artifactId>
-    </dependency>
     <dependency>
       <groupId>commons-io</groupId>
       <artifactId>commons-io</artifactId>
@@ -92,6 +88,10 @@
       <groupId>jakarta.xml.bind</groupId>
       <artifactId>jakarta.xml.bind-api</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-collections4</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-lang3</artifactId>

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/StringUtils.java

@@ -97,4 +97,30 @@ public static String bytes2String(byte[] bytes) {
   public static byte[] string2Bytes(String str) {
     return str.getBytes(UTF8);
   }
+
+  public static String getLexicographicallyLowerString(String val) {
+    if (val == null || val.isEmpty()) {
+      throw new IllegalArgumentException("Input string must not be null or empty");
+    }
+    char[] charVal = val.toCharArray();
+    int lastIdx = charVal.length - 1;
+    if (charVal[lastIdx] == Character.MIN_VALUE) {
+      throw new IllegalArgumentException("Cannot decrement character below Character.MIN_VALUE");
+    }
+    charVal[lastIdx] -= 1;
+    return String.valueOf(charVal);
+  }
+
+  public static String getLexicographicallyHigherString(String val) {
+    if (val == null || val.isEmpty()) {
+      throw new IllegalArgumentException("Input string must not be null or empty");
+    }
+    char[] charVal = val.toCharArray();
+    int lastIdx = charVal.length - 1;
+    if (charVal[lastIdx] == Character.MAX_VALUE) {
+      throw new IllegalArgumentException("Cannot increment character above Character.MAX_VALUE");
+    }
+    charVal[lastIdx] += 1;
+    return String.valueOf(charVal);
+  }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ratis/RatisHelper.java

@@ -544,6 +544,12 @@ public static void transferRatisLeadership(ConfigurationSource conf,
           + group.getPeers().stream().map(RaftPeer::getId)
               .collect(Collectors.toList()) + ".");
     }
+    if (!group.getPeer(targetPeerId).getStartupRole().equals(RaftProtos.RaftPeerRole.FOLLOWER)) {
+      throw new IOException("Target " + targetPeerId + " not in FOLLOWER role. "
+          + group.getPeers().stream()
+          .map(p -> p.getId() + ":" + p.getStartupRole())
+          .collect(Collectors.toList()) + ".");
+    }
 
     LOG.info("Start transferring leadership to {}", targetPeerId);
     try (RaftClient client = newRaftClient(SupportedRpcType.GRPC, null,
@@ -566,13 +572,19 @@ null, group, createRetryPolicy(conf), tlsConfig, conf)) {
       RaftClientReply setConf = null;
       try {
         // Set priority
-        final List<RaftPeer> peersWithNewPriorities = group.getPeers().stream()
+        final List<RaftPeer> followerWithNewPriorities = group.getPeers().stream()
+            .filter(peer -> peer.getStartupRole().equals(RaftProtos.RaftPeerRole.FOLLOWER))
+            .map(peer -> newRaftPeer(peer, targetPeerId))
+            .collect(Collectors.toList());
+        final List<RaftPeer> listenerWithNewPriorities = group.getPeers().stream()
+            .filter(peer -> peer.getStartupRole().equals(RaftProtos.RaftPeerRole.LISTENER))
             .map(peer -> newRaftPeer(peer, targetPeerId))
             .collect(Collectors.toList());
         // Set new configuration
-        setConf = client.admin().setConfiguration(peersWithNewPriorities);
+        setConf = client.admin().setConfiguration(followerWithNewPriorities, listenerWithNewPriorities);
         if (setConf.isSuccess()) {
-          LOG.info("Successfully set priority: {}", peersWithNewPriorities);
+          LOG.info("Successfully set priority: Follower: {}, Listener: {}", followerWithNewPriorities,
+              listenerWithNewPriorities);
         } else {
           throw new IOException("Failed to set priority.",
               setConf.getException());
@@ -598,13 +610,19 @@ null, group, createRetryPolicy(conf), tlsConfig, conf)) {
   }
 
   private static void resetPriorities(RaftGroup original, RaftClient client) {
-    final List<RaftPeer> resetPeers = original.getPeers().stream()
+    final List<RaftPeer> resetFollower = original.getPeers().stream()
+        .filter(peer -> peer.getStartupRole().equals(RaftProtos.RaftPeerRole.FOLLOWER))
+        .map(originalPeer -> RaftPeer.newBuilder(originalPeer)
+            .setPriority(NEUTRAL_PRIORITY).build())
+        .collect(Collectors.toList());
+    final List<RaftPeer> resetListener = original.getPeers().stream()
+        .filter(peer -> peer.getStartupRole().equals(RaftProtos.RaftPeerRole.LISTENER))
         .map(originalPeer -> RaftPeer.newBuilder(originalPeer)
             .setPriority(NEUTRAL_PRIORITY).build())
         .collect(Collectors.toList());
-    LOG.info("Resetting Raft peers priorities to {}", resetPeers);
+    LOG.info("Resetting Raft peers priorities to Follower: {}, Listener: {}", resetFollower, resetListener);
     try {
-      RaftClientReply reply = client.admin().setConfiguration(resetPeers);
+      RaftClientReply reply = client.admin().setConfiguration(resetFollower, resetListener);
       if (reply.isSuccess()) {
         LOG.info("Successfully reset priorities: {}", original);
       } else {

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

@@ -17,16 +17,11 @@
 
 package org.apache.hadoop.hdds.scm.container;
 
-import static org.apache.hadoop.hdds.HddsUtils.checksumToString;
-
-import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.databind.JsonSerializer;
-import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import java.io.IOException;
 import java.util.UUID;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.server.JsonUtils;
 
 /**
  * Class which stores ContainerReplica details on the client.
@@ -41,7 +36,7 @@ public final class ContainerReplicaInfo {
   private long keyCount;
   private long bytesUsed;
   private int replicaIndex = -1;
-  @JsonSerialize(using = LongToHexJsonSerializer.class)
+  @JsonSerialize(using = JsonUtils.ChecksumSerializer.class)
   private long dataChecksum;
 
   public static ContainerReplicaInfo fromProto(
@@ -100,13 +95,6 @@ public long getDataChecksum() {
     return dataChecksum;
   }
 
-  private static class LongToHexJsonSerializer extends JsonSerializer<Long> {
-    @Override
-    public void serialize(Long value, JsonGenerator gen, SerializerProvider provider) throws IOException {
-      gen.writeString(checksumToString(value));
-    }
-  }
-
   /**
    * Builder for ContainerReplicaInfo class.
    */

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

@@ -21,7 +21,7 @@
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
-import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

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

@@ -36,7 +36,7 @@
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Consumer;
-import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;