added integration test

Author: Gargi Jaiswal

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

@@ -148,6 +148,7 @@ private static void deleteFileQuietely(Path tarFilePath) {
 
   HddsVolume chooseNextVolume(long spaceToReserve) throws IOException {
     // Choose volume that can hold both container in tmp and dest directory
+    LOG.info("Choosing volume to reserve space : {}", spaceToReserve);
     return volumeChoosingPolicy.chooseVolume(
         StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()),
         spaceToReserve);

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

@@ -65,6 +65,8 @@ public OutputStream startUpload(long containerId, DatanodeDetails target,
     Long containerSize = null;
     Container<?> container = containerController.getContainer(containerId);
     if (container != null) {
+      LOG.info("Starting upload of container {} to {} with size {}",
+          containerId, target, container.getContainerData().getBytesUsed());
       containerSize = container.getContainerData().getBytesUsed();
     }
 

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerReplication.java

@@ -24,6 +24,7 @@
 import static org.apache.hadoop.hdds.scm.pipeline.MockPipeline.createPipeline;
 import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls.createContainer;
 import static org.apache.ozone.test.GenericTestUtils.waitFor;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import com.google.common.collect.ImmutableList;
 import java.io.IOException;
@@ -34,24 +35,36 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.ToLongFunction;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;
 import org.apache.hadoop.hdds.scm.container.replication.ReplicationManager.ReplicationManagerConfiguration;
 import org.apache.hadoop.hdds.utils.IOUtils;
 import org.apache.hadoop.ozone.HddsDatanodeService;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.container.ContainerTestHelper;
+import org.apache.hadoop.ozone.container.common.interfaces.Container;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine;
 import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
 import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.ozone.test.GenericTestUtils.LogCapturer;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.MethodSource;
 
 /**
  * Tests ozone containers replication.
@@ -157,6 +170,68 @@ void pushUnknownContainer() throws Exception {
         ReplicationSupervisor::getReplicationFailureCount);
   }
 
+  /**
+   * Provides stream of different container sizes for tests.
+   */
+  public static Stream<Arguments> sizeProvider() {
+    return Stream.of(
+        Arguments.of("Normal 2MB", 2L * 1024L * 1024L),
+        Arguments.of("Overallocated 6MB", 6L * 1024L * 1024L)
+    );
+  }
+
+  /**
+   * Tests push replication of a container with over-allocated size.
+   * The target datanode will need to reserve double the container size,
+   * which is greater than the configured max container size.
+   */
+  @ParameterizedTest(name = "for {0}")
+  @MethodSource("sizeProvider")
+  void testPushWithOverAllocatedContainer(String testName, Long containerSize)
+      throws Exception {
+    LogCapturer grpcLog = LogCapturer.captureLogs(GrpcContainerUploader.class);
+    LogCapturer containerImporterLog = LogCapturer.captureLogs(ContainerImporter.class);
+
+    DatanodeDetails source = cluster.getHddsDatanodes().get(0)
+        .getDatanodeDetails();
+
+    long containerID = createOverAllocatedContainer(source, containerSize);
+
+    DatanodeDetails target = selectOtherNode(source);
+    
+    // Get the original container size from source
+    Container<?> sourceContainer = getContainer(source, containerID);
+    long originalSize = sourceContainer.getContainerData().getBytesUsed();
+    
+    // Verify container is created with expected size
+    assertEquals(originalSize, containerSize);
+    
+    // Create replication command to push container to target
+    ReplicateContainerCommand cmd =
+        ReplicateContainerCommand.toTarget(containerID, target);
+
+    // Execute push replication
+    queueAndWaitForCompletion(cmd, source,
+        ReplicationSupervisor::getReplicationSuccessCount);
+
+    GenericTestUtils.waitFor(() -> {
+      String grpcLogs = grpcLog.getOutput();
+      String containerImporterLogOutput = containerImporterLog.getOutput();
+
+      return grpcLogs.contains("Starting upload of container " +
+          containerID + " to " + target + " with size " + originalSize) &&
+          containerImporterLogOutput.contains("Choosing volume to reserve space: " +
+              originalSize * 2);
+    }, 100, 1000);
+
+    // Verify container was successfully replicated to target
+    Container<?> targetContainer = getContainer(target, containerID);
+    long replicatedSize = targetContainer.getContainerData().getBytesUsed();
+
+    // verify sizes match exactly
+    assertEquals(originalSize, replicatedSize);
+  }
+
   /**
    * Queues {@code cmd} in {@code dn}'s state machine, and waits until the
    * command is completed, as indicated by {@code counter} having been
@@ -194,6 +269,8 @@ private static OzoneConfiguration createConfiguration() {
     OzoneConfiguration conf = new OzoneConfiguration();
     conf.setTimeDuration(OZONE_SCM_STALENODE_INTERVAL, 3, TimeUnit.SECONDS);
     conf.setTimeDuration(OZONE_SCM_DEADNODE_INTERVAL, 6, TimeUnit.SECONDS);
+    conf.setStorageSize(ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE, 5, StorageUnit.MB);
+    conf.setStorageSize(OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE, 1, StorageUnit.MB);
 
     ReplicationManagerConfiguration repConf =
         conf.getObject(ReplicationManagerConfiguration.class);
@@ -212,4 +289,71 @@ private static long createNewClosedContainer(DatanodeDetails dn)
     }
   }
 
+  private static long createOverAllocatedContainer(DatanodeDetails dn, Long targetDataSize) throws Exception {
+    long containerID = CONTAINER_ID.incrementAndGet();
+    try (XceiverClientSpi client = clientFactory.acquireClient(
+        createPipeline(singleton(dn)))) {
+      
+      // Create the container
+      createContainer(client, containerID, null);
+      
+      int chunkSize = 1 * 1024 * 1024; // 1MB chunks
+      long totalBytesWritten = 0;
+
+      // Write data in chunks until we reach target size
+      while (totalBytesWritten < targetDataSize) {
+        BlockID blockID = ContainerTestHelper.getTestBlockID(containerID);
+        
+        // Calculate remaining bytes and adjust chunk size if needed
+        long remainingBytes = targetDataSize - totalBytesWritten;
+        int currentChunkSize = (int) Math.min(chunkSize, remainingBytes);
+        
+        // Create a write chunk request with current chunk size
+        ContainerProtos.ContainerCommandRequestProto writeChunkRequest =
+            ContainerTestHelper.getWriteChunkRequest(
+                createPipeline(singleton(dn)), blockID, currentChunkSize);
+        
+        // Send write chunk command
+        client.sendCommand(writeChunkRequest);
+        
+        // Create and send put block command
+        ContainerProtos.ContainerCommandRequestProto putBlockRequest =
+            ContainerTestHelper.getPutBlockRequest(writeChunkRequest);
+        client.sendCommand(putBlockRequest);
+        
+        totalBytesWritten += currentChunkSize;
+      }
+      
+      // Close the container
+      ContainerProtos.CloseContainerRequestProto closeRequest =
+          ContainerProtos.CloseContainerRequestProto.newBuilder().build();
+      ContainerProtos.ContainerCommandRequestProto closeContainerRequest =
+          ContainerProtos.ContainerCommandRequestProto.newBuilder()
+              .setCmdType(ContainerProtos.Type.CloseContainer)
+              .setContainerID(containerID)
+              .setCloseContainer(closeRequest)
+              .setDatanodeUuid(dn.getUuidString())
+              .build();
+      client.sendCommand(closeContainerRequest);
+      
+      return containerID;
+    }
+  }
+
+  /**
+   * Gets the container from the specified datanode.
+   */
+  private Container<?> getContainer(DatanodeDetails datanode, long containerID) {
+    for (HddsDatanodeService datanodeService : cluster.getHddsDatanodes()) {
+      if (datanode.equals(datanodeService.getDatanodeDetails())) {
+        Container<?> container = datanodeService.getDatanodeStateMachine().getContainer()
+            .getContainerSet().getContainer(containerID);
+        if (container != null) {
+          return container;
+        }
+      }
+    }
+    throw new AssertionError("Container " + containerID + " not found on " + datanode);
+  }
+
 }