HDDS-16012. Detect replicas with equal BCSID and mismatched data checksums

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

@@ -107,6 +107,13 @@ public enum ContainerHealthState {
       "Containers in OPEN state without any healthy Pipeline",
       "OpenContainersWithoutPipeline"),
 
+  /**
+   * Replicas with the same BCSID have different data checksums.
+   */
+  DATA_CHECKSUM_MISMATCH((short) 10,
+      "Containers with replicas reporting the same BCSID but different data checksums",
+      "DataChecksumMismatchContainers"),
+
   // ========== Actual Combinations Found in Code (100+) ==========
 
   /**

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReplicaChecksumMismatch.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.container;
+
+import java.util.Collection;
+import java.util.Objects;
+import java.util.function.Function;
+import java.util.function.ToLongFunction;
+
+/**
+ * Detects different data checksums reported for the same container replica
+ * sequence ID.
+ */
+public final class ContainerReplicaChecksumMismatch {
+
+  private ContainerReplicaChecksumMismatch() {
+  }
+
+  /**
+   * Returns true when replicas with the same sequence ID report different
+   * non-zero data checksums. If any replica has not reported a sequence ID or
+   * checksum yet, no comparison is made.
+   */
+  public static <T> boolean hasMismatch(Collection<T> replicas,
+      Function<T, Long> sequenceId, ToLongFunction<T> dataChecksum) {
+    Objects.requireNonNull(sequenceId, "sequenceId == null");
+    Objects.requireNonNull(dataChecksum, "dataChecksum == null");
+
+    if (replicas == null || replicas.size() < 2) {
+      return false;
+    }
+
+    for (T replica : replicas) {
+      Long replicaSequenceId = sequenceId.apply(replica);
+      long replicaDataChecksum = dataChecksum.applyAsLong(replica);
+      if (replicaSequenceId == null || replicaDataChecksum == 0) {
+        return false;
+      }
+    }
+
+    for (T left : replicas) {
+      Long leftSequenceId = sequenceId.apply(left);
+      long leftDataChecksum = dataChecksum.applyAsLong(left);
+      for (T right : replicas) {
+        if (leftSequenceId.equals(sequenceId.apply(right)) &&
+            leftDataChecksum != dataChecksum.applyAsLong(right)) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * 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.container;
+
+import java.util.List;
+
+/**
+ * Container replica information and the checksum mismatch state determined
+ * by SCM.
+ */
+public final class ContainerReplicaInfoResult {
+
+  private final List<ContainerReplicaInfo> replicas;
+  private final boolean dataChecksumMismatch;
+
+  public ContainerReplicaInfoResult(List<ContainerReplicaInfo> replicas,
+      boolean dataChecksumMismatch) {
+    this.replicas = replicas;
+    this.dataChecksumMismatch = dataChecksumMismatch;
+  }
+
+  public List<ContainerReplicaInfo> getReplicas() {
+    return replicas;
+  }
+
+  public boolean hasDataChecksumMismatch() {
+    return dataChecksumMismatch;
+  }
+}

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

@@ -90,6 +90,15 @@ public void incrementAndSample(ContainerHealthState stat, ContainerInfo containe
     containerHealthState = stat;
   }
 
+  /**
+   * Increments a health state and records the container ID without changing
+   * the primary health state of the container currently being processed.
+   */
+  public void incrementAndSampleAdditionalState(
+      ContainerHealthState stat, ContainerID containerID) {
+    incrementAndSample(stat.name(), containerID);
+  }
+
   public void increment(HddsProtos.LifeCycleState stat) {
     increment(stat.toString());
   }

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/TestContainerHealthState.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.hdds.scm.container;
 
+import static org.apache.hadoop.hdds.scm.container.ContainerHealthState.DATA_CHECKSUM_MISMATCH;
 import static org.apache.hadoop.hdds.scm.container.ContainerHealthState.EMPTY;
 import static org.apache.hadoop.hdds.scm.container.ContainerHealthState.HEALTHY;
 import static org.apache.hadoop.hdds.scm.container.ContainerHealthState.MISSING;
@@ -59,6 +60,7 @@ public void testIndividualStateValues() {
     assertEquals(7, OPEN_UNHEALTHY.getValue());
     assertEquals(8, QUASI_CLOSED_STUCK.getValue());
     assertEquals(9, OPEN_WITHOUT_PIPELINE.getValue());
+    assertEquals(10, DATA_CHECKSUM_MISMATCH.getValue());
   }
 
   @Test
@@ -101,6 +103,7 @@ public void testFromValueIndividualStates() {
     assertEquals(OPEN_UNHEALTHY, ContainerHealthState.fromValue((short) 7));
     assertEquals(QUASI_CLOSED_STUCK, ContainerHealthState.fromValue((short) 8));
     assertEquals(OPEN_WITHOUT_PIPELINE, ContainerHealthState.fromValue((short) 9));
+    assertEquals(DATA_CHECKSUM_MISMATCH, ContainerHealthState.fromValue((short) 10));
   }
 
   @Test
@@ -136,11 +139,11 @@ public void testAllEnumValuesAreUnique() {
 
   @Test
   public void testIndividualStateCount() {
-    // Should have 10 individual states (0-9)
+    // Should have 11 individual states (0-10)
     long individualCount = java.util.Arrays.stream(ContainerHealthState.values())
         .filter(s -> s.getValue() >= 0 && s.getValue() <= 99)
         .count();
-    assertEquals(10, individualCount, "Expected 10 individual states");
+    assertEquals(11, individualCount, "Expected 11 individual states");
   }
 
   @Test
@@ -154,10 +157,10 @@ public void testCombinationStateCount() {
 
   @Test
   public void testNoGapsInIndividualValues() {
-    // Individual states should be sequential: 0-9
-    for (short i = 0; i <= 9; i++) {
+    // Individual states should be sequential: 0-10
+    for (short i = 0; i <= 10; i++) {
       ContainerHealthState state = ContainerHealthState.fromValue(i);
-      assertTrue(state.getValue() >= 0 && state.getValue() <= 9,
+      assertTrue(state.getValue() >= 0 && state.getValue() <= 10,
           "Value " + i + " should map to an individual state");
     }
   }

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/TestContainerReplicaChecksumMismatch.java

@@ -0,0 +1,117 @@
+/*
+ * 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.container;
+
+import static org.apache.hadoop.hdds.scm.container.ContainerReplicaChecksumMismatch.hasMismatch;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class TestContainerReplicaChecksumMismatch {
+
+  @Test
+  void detectsDifferentChecksumsAtTheSameSequenceId() {
+    List<Replica> replicas = Arrays.asList(
+        new Replica(10L, 100L),
+        new Replica(10L, 200L),
+        new Replica(10L, 100L));
+
+    assertTrue(hasMismatch(replicas, Replica::getSequenceId,
+        Replica::getDataChecksum));
+  }
+
+  @Test
+  void ignoresDifferentChecksumsAtDifferentSequenceIds() {
+    List<Replica> replicas = Arrays.asList(
+        new Replica(10L, 100L),
+        new Replica(11L, 200L));
+
+    assertFalse(hasMismatch(replicas, Replica::getSequenceId,
+        Replica::getDataChecksum));
+  }
+
+  @Test
+  void ignoresEqualChecksumsAtDifferentSequenceIds() {
+    List<Replica> replicas = Arrays.asList(
+        new Replica(10L, 100L),
+        new Replica(11L, 100L));
+
+    assertFalse(hasMismatch(replicas, Replica::getSequenceId,
+        Replica::getDataChecksum));
+  }
+
+  @Test
+  void detectsMismatchWithinOneSequenceIdGroup() {
+    List<Replica> replicas = Arrays.asList(
+        new Replica(10L, 100L),
+        new Replica(11L, 200L),
+        new Replica(11L, 300L));
+
+    assertTrue(hasMismatch(replicas, Replica::getSequenceId,
+        Replica::getDataChecksum));
+  }
+
+  @Test
+  void waitsUntilEveryReplicaReportsADataChecksum() {
+    List<Replica> replicas = Arrays.asList(
+        new Replica(10L, 100L),
+        new Replica(10L, 200L),
+        new Replica(10L, 0L));
+
+    assertFalse(hasMismatch(replicas, Replica::getSequenceId,
+        Replica::getDataChecksum));
+  }
+
+  @Test
+  void waitsUntilEveryReplicaReportsASequenceId() {
+    List<Replica> replicas = Arrays.asList(
+        new Replica(10L, 100L),
+        new Replica(null, 200L));
+
+    assertFalse(hasMismatch(replicas, Replica::getSequenceId,
+        Replica::getDataChecksum));
+  }
+
+  @Test
+  void requiresAtLeastTwoReplicas() {
+    assertFalse(hasMismatch(Collections.singletonList(new Replica(10L, 100L)),
+        Replica::getSequenceId, Replica::getDataChecksum));
+  }
+
+  private static final class Replica {
+    private final Long sequenceId;
+    private final long dataChecksum;
+
+    private Replica(Long sequenceId, long dataChecksum) {
+      this.sequenceId = sequenceId;
+      this.dataChecksum = dataChecksum;
+    }
+
+    private Long getSequenceId() {
+      return sequenceId;
+    }
+
+    private long getDataChecksum() {
+      return dataChecksum;
+    }
+  }
+}

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/TestReplicationManagerReport.java

@@ -28,6 +28,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.ThreadLocalRandom;
@@ -115,6 +116,7 @@ void testJsonOutput() throws IOException {
     assertEquals(0, stats.get("OPEN_UNHEALTHY").longValue());
     assertEquals(0, stats.get("QUASI_CLOSED_STUCK").longValue());
     assertEquals(0, stats.get("OPEN_WITHOUT_PIPELINE").longValue());
+    assertEquals(0, stats.get("DATA_CHECKSUM_MISMATCH").longValue());
 
     JsonNode samples = json.get("samples");
     assertEquals(ARRAY, samples.get("UNDER_REPLICATED").getNodeType());
@@ -137,6 +139,20 @@ void testContainerIDsCanBeSampled() {
         report.getStat(ContainerHealthState.MIS_REPLICATED));
   }
 
+  @Test
+  void testSampleByContainerIDDoesNotChangeContainerHealthState() {
+    ContainerID containerID = ContainerID.valueOf(1);
+    report.incrementAndSampleAdditionalState(
+        ContainerHealthState.DATA_CHECKSUM_MISMATCH, containerID);
+
+    assertEquals(1,
+        report.getStat(ContainerHealthState.DATA_CHECKSUM_MISMATCH));
+    assertEquals(Collections.singletonList(containerID),
+        report.getSample(ContainerHealthState.DATA_CHECKSUM_MISMATCH));
+    assertEquals(ContainerHealthState.HEALTHY,
+        report.getContainerHealthState());
+  }
+
   @Test
   void testSamplesAreLimited() {
     verifySampleLimit(report, 100);

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

@@ -38,6 +38,7 @@
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerListResult;
 import org.apache.hadoop.hdds.scm.container.ContainerReplicaInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerReplicaInfoResult;
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
@@ -83,6 +84,17 @@ ContainerWithPipeline getContainerWithPipeline(long containerId)
   List<ContainerReplicaInfo> getContainerReplicas(
       long containerId) throws IOException;
 
+  /**
+   * Gets replica information and the container-level status determined by
+   * SCM.
+   *
+   * @param containerId the container ID
+   * @return replicas and container-level status
+   * @throws IOException on failure
+   */
+  ContainerReplicaInfoResult getContainerReplicasWithStatus(long containerId)
+      throws IOException;
+
   /**
    * Close a container.
    *