WIP migrate reconciliation unit tests

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java

@@ -152,8 +152,6 @@ public class TestKeyValueHandler {
 
   private static final long DUMMY_CONTAINER_ID = 9999;
   private static final String DUMMY_PATH = "dummy/dir/doesnt/exist";
-  private static final int CHUNK_LEN = 3 * (int) OzoneConsts.KB;
-  private static final int CHUNKS_PER_BLOCK = 4;
   private static final String DATANODE_UUID = UUID.randomUUID().toString();
   private static final String CLUSTER_ID = UUID.randomUUID().toString();
 
@@ -162,26 +160,6 @@ public class TestKeyValueHandler {
   private OzoneConfiguration conf;
   private ContainerSet mockContainerSet;
 
-  /**
-   * Number of corrupt blocks and chunks.
-   */
-  public static Stream<Arguments> corruptionValues() {
-    return Stream.of(
-        Arguments.of(5, 0),
-        Arguments.of(0, 5),
-        Arguments.of(0, 10),
-        Arguments.of(10, 0),
-        Arguments.of(5, 10),
-        Arguments.of(10, 5),
-        Arguments.of(2, 3),
-        Arguments.of(3, 2),
-        Arguments.of(4, 6),
-        Arguments.of(6, 4),
-        Arguments.of(6, 9),
-        Arguments.of(9, 6)
-    );
-  }
-
   @BeforeEach
   public void setup() throws IOException {
     // Create mock HddsDispatcher and KeyValueHandler.
@@ -204,7 +182,6 @@ public void setup() throws IOException {
         mock(ContainerMetrics.class),
         mock(TokenVerifier.class)
     );
-
   }
 
   /**
@@ -593,151 +570,6 @@ public void testContainerChecksumInvocation(ContainerLayoutVersion layoutVersion
     Assertions.assertEquals(1, icrCount.get());
   }
 
-  @ParameterizedTest
-  @MethodSource("corruptionValues")
-  public void testFullContainerReconciliation(int numBlocks, int numChunks) throws Exception {
-    KeyValueHandler kvHandler = createKeyValueHandler(tempDir);
-    ContainerChecksumTreeManager checksumManager = kvHandler.getChecksumManager();
-
-    ContainerController controller = new ContainerController(mockContainerSet,
-        Collections.singletonMap(ContainerType.KeyValueContainer, kvHandler));
-    OnDemandContainerDataScanner.init(conf.getObject(ContainerScannerConfiguration.class), controller, checksumManager);
-
-    DNContainerOperationClient dnClient = new DNContainerOperationClient(conf, null, null);
-    final long containerID = 100L;
-    // Create 3 containers with 15 blocks each and 3 replicas.
-    List<KeyValueContainer> containers = createContainerWithBlocks(kvHandler, containerID, 15, 3);
-    assertEquals(3, containers.size());
-
-    // Introduce corruption in each container on different replicas.
-    introduceCorruption(kvHandler, containers.get(1), numBlocks, numChunks, false);
-    introduceCorruption(kvHandler, containers.get(2), numBlocks, numChunks, true);
-    // Use synchronous on-demand scans to re-build the merkle trees after corruption.
-    waitForContainerScans(containers);
-
-    // Without reconciliation, checksums should be different because of the corruption.
-    Set<Long> checksumsBeforeReconciliation = new HashSet<>();
-    for (KeyValueContainer kvContainer : containers) {
-      Optional<ContainerProtos.ContainerChecksumInfo> containerChecksumInfo =
-          checksumManager.read(kvContainer.getContainerData());
-      assertTrue(containerChecksumInfo.isPresent());
-      long dataChecksum = containerChecksumInfo.get().getContainerMerkleTree().getDataChecksum();
-      assertEquals(kvContainer.getContainerData().getDataChecksum(), dataChecksum);
-      checksumsBeforeReconciliation.add(dataChecksum);
-    }
-    // There should be more than 1 checksum because of the corruption.
-    assertTrue(checksumsBeforeReconciliation.size() > 1);
-
-    List<DatanodeDetails> datanodes = ImmutableList.of(randomDatanodeDetails(), randomDatanodeDetails(),
-        randomDatanodeDetails());
-    Map<String, KeyValueContainer> dnToContainerMap = new HashMap<>();
-    dnToContainerMap.put(datanodes.get(0).getUuidString(), containers.get(0));
-    dnToContainerMap.put(datanodes.get(1).getUuidString(), containers.get(1));
-    dnToContainerMap.put(datanodes.get(2).getUuidString(), containers.get(2));
-
-    // Setup mock for each datanode network calls needed for reconciliation.
-    try (MockedStatic<ContainerProtocolCalls> containerProtocolMock =
-             Mockito.mockStatic(ContainerProtocolCalls.class)) {
-      mockContainerProtocolCalls(containerProtocolMock, dnToContainerMap, checksumManager, kvHandler, containerID);
-
-      kvHandler.reconcileContainer(dnClient, containers.get(0), datanodes);
-      kvHandler.reconcileContainer(dnClient, containers.get(1), datanodes);
-      kvHandler.reconcileContainer(dnClient, containers.get(2), datanodes);
-
-      // After reconciliation, checksums should be the same for all containers.
-      // Reconciliation should have updated the tree based on the updated metadata that was obtained for the
-      // previously corrupted data. We do not need to wait for the full data scan to complete.
-      ContainerProtos.ContainerChecksumInfo prevContainerChecksumInfo = null;
-      for (KeyValueContainer kvContainer : containers) {
-        kvHandler.createContainerMerkleTreeFromMetadata(kvContainer);
-        Optional<ContainerProtos.ContainerChecksumInfo> containerChecksumInfo =
-            checksumManager.read(kvContainer.getContainerData());
-        assertTrue(containerChecksumInfo.isPresent());
-        long dataChecksum = containerChecksumInfo.get().getContainerMerkleTree().getDataChecksum();
-        assertEquals(kvContainer.getContainerData().getDataChecksum(), dataChecksum);
-        if (prevContainerChecksumInfo != null) {
-          assertEquals(prevContainerChecksumInfo.getContainerMerkleTree().getDataChecksum(), dataChecksum);
-        }
-        prevContainerChecksumInfo = containerChecksumInfo.get();
-      }
-    }
-  }
-
-  public void waitForContainerScans(List<KeyValueContainer> containers) throws Exception {
-    for (KeyValueContainer container: containers) {
-      // The on-demand scanner has been initialized to pull from the mock container set.
-      // Make it pull the corresponding container instance to scan in this run based on ID.
-      long containerID = container.getContainerData().getContainerID();
-      Mockito.doReturn(container).when(mockContainerSet).getContainer(containerID);
-
-      Optional<Future<?>> scanFuture = OnDemandContainerDataScanner.scanContainer(container);
-      assertTrue(scanFuture.isPresent());
-      // Wait for on-demand scan to complete.
-      scanFuture.get().get();
-    }
-  }
-
-  private void mockContainerProtocolCalls(MockedStatic<ContainerProtocolCalls> containerProtocolMock,
-                                          Map<String, KeyValueContainer> dnToContainerMap,
-                                          ContainerChecksumTreeManager checksumManager,
-                                          KeyValueHandler kvHandler,
-                                          long containerID) {
-    // Mock getContainerChecksumInfo
-    containerProtocolMock.when(() -> ContainerProtocolCalls.getContainerChecksumInfo(any(), anyLong(), any()))
-        .thenAnswer(inv -> {
-          XceiverClientSpi xceiverClientSpi = inv.getArgument(0);
-          Pipeline pipeline = xceiverClientSpi.getPipeline();
-          assertEquals(1, pipeline.size());
-          DatanodeDetails dn = pipeline.getFirstNode();
-          KeyValueContainer container = dnToContainerMap.get(dn.getUuidString());
-          ByteString checksumInfo = checksumManager.getContainerChecksumInfo(container.getContainerData());
-          return ContainerProtos.GetContainerChecksumInfoResponseProto.newBuilder()
-              .setContainerID(containerID)
-              .setContainerChecksumInfo(checksumInfo)
-              .build();
-        });
-
-    // Mock getBlock
-    containerProtocolMock.when(() -> ContainerProtocolCalls.getBlock(any(), any(), any(), any(), anyMap()))
-        .thenAnswer(inv -> {
-          XceiverClientSpi xceiverClientSpi = inv.getArgument(0);
-          Pipeline pipeline = xceiverClientSpi.getPipeline();
-          assertEquals(1, pipeline.size());
-          DatanodeDetails dn = pipeline.getFirstNode();
-          KeyValueContainer container = dnToContainerMap.get(dn.getUuidString());
-          ContainerProtos.BlockData blockData = kvHandler.getBlockManager().getBlock(container, inv.getArgument(2))
-                .getProtoBufMessage();
-          return ContainerProtos.GetBlockResponseProto.newBuilder()
-              .setBlockData(blockData)
-              .build();
-        });
-
-    // Mock readChunk
-    containerProtocolMock.when(() -> ContainerProtocolCalls.readChunk(any(), any(), any(), any(), any()))
-        .thenAnswer(inv -> {
-          XceiverClientSpi xceiverClientSpi = inv.getArgument(0);
-          Pipeline pipeline = xceiverClientSpi.getPipeline();
-          assertEquals(1, pipeline.size());
-          DatanodeDetails dn = pipeline.getFirstNode();
-          KeyValueContainer container = dnToContainerMap.get(dn.getUuidString());
-          return createReadChunkResponse(inv, container, kvHandler);
-        });
-  }
-
-  // Helper method to create readChunk responses
-  private ContainerProtos.ReadChunkResponseProto createReadChunkResponse(InvocationOnMock inv,
-                                                                         KeyValueContainer container,
-                                                                         KeyValueHandler kvHandler) throws IOException {
-    ContainerProtos.DatanodeBlockID blockId = inv.getArgument(2);
-    ContainerProtos.ChunkInfo chunkInfo = inv.getArgument(1);
-    return ContainerProtos.ReadChunkResponseProto.newBuilder()
-        .setBlockID(blockId)
-        .setChunkData(chunkInfo)
-        .setData(kvHandler.getChunkManager().readChunk(container, BlockID.getFromProtobuf(blockId),
-                ChunkInfo.getFromProtoBuf(chunkInfo), null).toByteString())
-        .build();
-  }
-
   @Test
   public void testGetContainerChecksumInfoOnInvalidContainerStates() {
     when(handler.handleGetContainerChecksumInfo(any(), any())).thenCallRealMethod();
@@ -842,6 +674,7 @@ private static ContainerCommandRequestProto createContainerRequest(
 
   private KeyValueHandler createKeyValueHandler(Path path) throws IOException {
     final ContainerSet containerSet = new ContainerSet(1000);
+
     final MutableVolumeSet volumeSet = mock(MutableVolumeSet.class);
 
     HddsVolume hddsVolume = new HddsVolume.Builder(path.toString()).conf(conf)
@@ -859,165 +692,14 @@ private KeyValueHandler createKeyValueHandler(Path path) throws IOException {
     hddsVolume.getVolumeInfoStats().unregister();
     hddsVolume.getVolumeIOStats().unregister();
     ContainerMetrics.remove();
-    return kvHandler;
-  }
 
-  /**
-   * Creates a container with normal and deleted blocks.
-   * First it will insert normal blocks, and then it will insert
-   * deleted blocks.
-   */
-  protected List<KeyValueContainer> createContainerWithBlocks(KeyValueHandler kvHandler, long containerId,
-                                                              int blocks, int numContainerCopy)
-      throws Exception {
-    String strBlock = "block";
-    String strChunk = "chunkFile";
-    List<KeyValueContainer> containers = new ArrayList<>();
-    MutableVolumeSet volumeSet = new MutableVolumeSet(DATANODE_UUID, conf, null,
-        StorageVolume.VolumeType.DATA_VOLUME, null);
-    createDbInstancesForTestIfNeeded(volumeSet, CLUSTER_ID, CLUSTER_ID, conf);
-    int bytesPerChecksum = 2 * (int) OzoneConsts.KB;
-    Checksum checksum = new Checksum(ContainerProtos.ChecksumType.SHA256,
-        bytesPerChecksum);
-    byte[] chunkData = RandomStringUtils.randomAscii(CHUNK_LEN).getBytes(UTF_8);
-    ChecksumData checksumData = checksum.computeChecksum(chunkData);
-
-    for (int j =  0; j < numContainerCopy; j++) {
-      KeyValueContainerData containerData = new KeyValueContainerData(containerId,
-          ContainerLayoutVersion.FILE_PER_BLOCK, (long) CHUNKS_PER_BLOCK * CHUNK_LEN * blocks,
-          UUID.randomUUID().toString(), UUID.randomUUID().toString());
-      Path kvContainerPath = Files.createDirectory(tempDir.resolve(containerId + "-" + j));
-      containerData.setMetadataPath(kvContainerPath.toString());
-      containerData.setDbFile(kvContainerPath.toFile());
-
-      KeyValueContainer container = new KeyValueContainer(containerData, conf);
-      StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList())
-          .forEach(hddsVolume -> hddsVolume.setDbParentDir(kvContainerPath.toFile()));
-      container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString());
-      assertNotNull(containerData.getChunksPath());
-      File chunksPath = new File(containerData.getChunksPath());
-      ContainerLayoutTestInfo.FILE_PER_BLOCK.validateFileCount(chunksPath, 0, 0);
-
-      List<ContainerProtos.ChunkInfo> chunkList = new ArrayList<>();
-      for (int i = 0; i < blocks; i++) {
-        BlockID blockID = new BlockID(containerId, i);
-        BlockData blockData = new BlockData(blockID);
-
-        chunkList.clear();
-        for (long chunkCount = 0; chunkCount < CHUNKS_PER_BLOCK; chunkCount++) {
-          String chunkName = strBlock + i + strChunk + chunkCount;
-          long offset = chunkCount * CHUNK_LEN;
-          ChunkInfo info = new ChunkInfo(chunkName, offset, CHUNK_LEN);
-          info.setChecksumData(checksumData);
-          chunkList.add(info.getProtoBufMessage());
-          kvHandler.getChunkManager().writeChunk(container, blockID, info,
-              ByteBuffer.wrap(chunkData), WRITE_STAGE);
-        }
-        kvHandler.getChunkManager().finishWriteChunks(container, blockData);
-        blockData.setChunks(chunkList);
-        blockData.setBlockCommitSequenceId(i);
-        kvHandler.getBlockManager().putBlock(container, blockData);
-      }
-
-      ContainerLayoutTestInfo.FILE_PER_BLOCK.validateFileCount(chunksPath, blocks, (long) blocks * CHUNKS_PER_BLOCK);
-      container.markContainerForClose();
-      kvHandler.closeContainer(container);
-      containers.add(container);
-    }
-
-    return containers;
-  }
-
-  /**
-   * Introduce corruption in the container.
-   * 1. Delete blocks from the container.
-   * 2. Corrupt chunks at an offset.
-   * If revers is true, the blocks and chunks are deleted in reverse order.
-   */
-  private void introduceCorruption(KeyValueHandler kvHandler, KeyValueContainer keyValueContainer, int numBlocks,
-                                   int numChunks, boolean reverse) throws IOException {
-    Random random = new Random();
-    KeyValueContainerData containerData = keyValueContainer.getContainerData();
-    // Simulate missing blocks
-    try (DBHandle handle = BlockUtils.getDB(containerData, conf);
-         BatchOperation batch = handle.getStore().getBatchHandler().initBatchOperation()) {
-      List<BlockData> blockDataList = kvHandler.getBlockManager().listBlock(keyValueContainer, -1, 100);
-      int size = blockDataList.size();
-      for (int i = 0; i < numBlocks; i++) {
-        BlockData blockData = reverse ? blockDataList.get(size - 1 - i) : blockDataList.get(i);
-        File blockFile = getBlock(keyValueContainer, blockData.getBlockID().getLocalID());
-        Assertions.assertTrue(blockFile.delete());
-        handle.getStore().getBlockDataTable().deleteWithBatch(batch, containerData.getBlockKey(blockData.getLocalID()));
-      }
-      handle.getStore().getBatchHandler().commitBatchOperation(batch);
-    }
-//    Files.deleteIfExists(getContainerChecksumFile(keyValueContainer.getContainerData()).toPath());
-//    kvHandler.createContainerMerkleTreeFromMetadata(keyValueContainer);
-
-    // Corrupt chunks at an offset.
-    List<BlockData> blockDataList = kvHandler.getBlockManager().listBlock(keyValueContainer, -1, 100);
-    int size = blockDataList.size();
-    for (int i = 0; i < numChunks; i++) {
-      int blockIndex = reverse ? size - 1 - (i % size) : i % size;
-      BlockData blockData = blockDataList.get(blockIndex);
-      int chunkIndex = i / size;
-      File blockFile = getBlock(keyValueContainer, blockData.getBlockID().getLocalID());
-      List<ContainerProtos.ChunkInfo> chunks = new ArrayList<>(blockData.getChunks());
-      ContainerProtos.ChunkInfo chunkInfo = chunks.remove(chunkIndex);
-      corruptFileAtOffset(blockFile, (int) chunkInfo.getOffset(), (int) chunkInfo.getLen());
-
-      // TODO: On-demand scanner (HDDS-10374) should detect this corruption and generate container merkle tree.
-//      ContainerProtos.ContainerChecksumInfo.Builder builder = kvHandler.getChecksumManager()
-//          .read(containerData).get().toBuilder();
-//      List<ContainerProtos.BlockMerkleTree> blockMerkleTreeList = builder.getContainerMerkleTree()
-//          .getBlockMerkleTreeList();
-//      assertEquals(size, blockMerkleTreeList.size());
-
-//      builder.getContainerMerkleTreeBuilder().clearBlockMerkleTree();
-//      for (int j = 0; j < blockMerkleTreeList.size(); j++) {
-//        ContainerProtos.BlockMerkleTree.Builder blockMerkleTreeBuilder = blockMerkleTreeList.get(j).toBuilder();
-//        if (j == blockIndex) {
-//          List<ContainerProtos.ChunkMerkleTree.Builder> chunkMerkleTreeBuilderList =
-//              blockMerkleTreeBuilder.getChunkMerkleTreeBuilderList();
-//          chunkMerkleTreeBuilderList.get(chunkIndex).setIsHealthy(false).setDataChecksum(random.nextLong());
-//          blockMerkleTreeBuilder.setDataChecksum(random.nextLong());
-//        }
-//        builder.getContainerMerkleTreeBuilder().addBlockMerkleTree(blockMerkleTreeBuilder.build());
-//      }
-//      builder.getContainerMerkleTreeBuilder().setDataChecksum(random.nextLong());
-//      Files.deleteIfExists(getContainerChecksumFile(keyValueContainer.getContainerData()).toPath());
-//      writeContainerDataTreeProto(keyValueContainer.getContainerData(), builder.getContainerMerkleTree());
-    }
-  }
-
-  /**
-   * Overwrite the file with random bytes at an offset within the given length.
-   */
-  public static void corruptFileAtOffset(File file, int offset, int chunkLength) {
-    try {
-      final int fileLength = (int) file.length();
-      assertTrue(fileLength >= offset + chunkLength);
-      final int chunkEnd = offset + chunkLength;
-
-      Path path = file.toPath();
-      final byte[] original = IOUtils.readFully(Files.newInputStream(path), fileLength);
-
-      // Corrupt the last byte and middle bytes of the block. The scanner should log this as two errors.
-      final byte[] corruptedBytes = Arrays.copyOf(original, fileLength);
-      corruptedBytes[chunkEnd - 1] = (byte) (original[chunkEnd - 1] << 1);
-      final long chunkMid = offset + ((long) chunkLength - offset) / 2;
-      corruptedBytes[(int) (chunkMid / 2)] = (byte) (original[(int) (chunkMid / 2)] << 1);
-
-
-      Files.write(path, corruptedBytes,
-          StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.SYNC);
+    // Register the on-demand container scanner with the container set used by the KeyValueHandler.
+    ContainerController controller = new ContainerController(containerSet,
+        Collections.singletonMap(ContainerType.KeyValueContainer, kvHandler));
+    OnDemandContainerDataScanner onDemandScanner = new OnDemandContainerDataScanner(
+        conf.getObject(ContainerScannerConfiguration.class), controller, kvHandler.getChecksumManager());
+    containerSet.registerContainerScanHandler(onDemandScanner::scanContainer);
 
-      assertThat(IOUtils.readFully(Files.newInputStream(path), fileLength))
-          .isEqualTo(corruptedBytes)
-          .isNotEqualTo(original);
-    } catch (IOException ex) {
-      // Fail the test.
-      throw new UncheckedIOException(ex);
-    }
+    return kvHandler;
   }
 }