HDDS-13891. Fixing failing tests.

Author: Devesh Kumar Singh

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -4414,6 +4414,20 @@
       Interval in MINUTES by Recon to request SCM DB Snapshot.
     </description>
   </property>
+
+  <property>
+    <name>ozone.recon.container.health.use.scm.report</name>
+    <value>false</value>
+    <tag>OZONE, MANAGEMENT, RECON</tag>
+    <description>
+      Feature flag to enable ContainerHealthTaskV2 which uses SCM's ReplicationManager
+      as the single source of truth for container health states. When enabled, V2 task
+      will sync container health states directly from SCM instead of computing them locally.
+      This provides more accurate and consistent container health reporting.
+      Default is false (uses legacy implementation).
+    </description>
+  </property>
+
   <property>
     <name>ozone.om.snapshot.compaction.dag.max.time.allowed</name>
     <value>30d</value>

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/fsck/ContainerHealthTaskV2.java

@@ -349,7 +349,7 @@ private UnhealthyContainerRecordV2 createRecord(
       int actualReplicaCount,
       String reason) {
 
-    int replicaDelta = actualReplicaCount - expectedReplicaCount;
+    int replicaDelta = expectedReplicaCount - actualReplicaCount;
 
     return new UnhealthyContainerRecordV2(
         container.getContainerID(),

hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTaskV2.java

@@ -22,6 +22,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -316,13 +317,24 @@ public void testContainerInSCMButNotInRecon() throws Exception {
     List<ContainerInfo> scmContainers = getMockContainers(1);
     ContainerInfo scmContainer = scmContainers.get(0);
 
-    when(scmClientMock.getListOfContainers(anyInt(), anyInt(),
-        any(HddsProtos.LifeCycleState.class))).thenReturn(scmContainers);
+    // Mock getListOfContainers to handle pagination correctly
+    // Return container for CLOSED state with startId=0, empty otherwise
+    when(scmClientMock.getListOfContainers(anyLong(), anyInt(),
+        any(HddsProtos.LifeCycleState.class))).thenAnswer(invocation -> {
+          long startId = invocation.getArgument(0);
+          HddsProtos.LifeCycleState state = invocation.getArgument(2);
+          // Only return container for CLOSED state and startId=0
+          if (state == HddsProtos.LifeCycleState.CLOSED && startId == 0) {
+            return scmContainers;
+          }
+          return Collections.emptyList();
+        });
     when(containerManagerMock.getContainer(scmContainer.containerID()))
         .thenThrow(new ContainerNotFoundException("Container not found in Recon"));
 
     ReplicationManagerReport report = createMockReport(0, 1, 0, 0);
-    when(scmClientMock.checkContainerStatus(scmContainer)).thenReturn(report);
+    // Use any() matcher since getListOfContainers returns a list with the same container
+    when(scmClientMock.checkContainerStatus(any(ContainerInfo.class))).thenReturn(report);
 
     ReconTaskConfig reconTaskConfig = new ReconTaskConfig();
     reconTaskConfig.setMissingContainerTaskInterval(Duration.ofSeconds(2));
@@ -378,7 +390,7 @@ public void testContainerInReconButNotInSCM() throws Exception {
     // SCM doesn't have this container
     when(scmClientMock.checkContainerStatus(reconContainer))
         .thenThrow(new ContainerNotFoundException("Container not found in SCM"));
-    when(scmClientMock.getListOfContainers(anyInt(), anyInt(),
+    when(scmClientMock.getListOfContainers(anyLong(), anyInt(),
         any(HddsProtos.LifeCycleState.class))).thenReturn(Collections.emptyList());
 
     // Insert a record for this container first

hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/AbstractReconSqlDBTest.java

@@ -179,7 +179,7 @@ public String getDriverClass() {
         @Override
         public String getJdbcUrl() {
           return "jdbc:derby:" + tempDir.getAbsolutePath() +
-              File.separator + "derby_recon.db";
+              File.separator + "derby_recon.db;user=RECON;create=true";
         }
 
         @Override