HDDS-15052. Revert HDDS-14040. Ozone client hang for data write in failure scenario (#10140)

Author: sumitagrawl

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java

@@ -582,8 +582,7 @@ private CompletableFuture<Message> writeStateMachineData(
     try {
       validateLongRunningWrite();
     } catch (StorageContainerException e) {
-      ContainerCommandResponseProto result = ContainerUtils.logAndReturnError(LOG, e, requestProto);
-      return CompletableFuture.completedFuture(result::toByteString);
+      return completeExceptionally(e);
     }
     final WriteChunkRequestProto write = requestProto.getWriteChunk();
     RaftServer server = ratisServer.getServer();
@@ -632,11 +631,8 @@ private CompletableFuture<Message> writeStateMachineData(
             // see the stateMachine is marked unhealthy by other parallel thread
             unhealthyContainers.add(write.getBlockID().getContainerID());
             stateMachineHealthy.set(false);
-            StorageContainerException sce = new StorageContainerException("Failed to write chunk data",
-                e, ContainerProtos.Result.CONTAINER_INTERNAL_ERROR);
-            ContainerCommandResponseProto result = ContainerUtils.logAndReturnError(LOG, sce, requestProto);
-            raftFuture.complete(result::toByteString);
-            return result;
+            raftFuture.completeExceptionally(e);
+            throw e;
           } finally {
             // Remove the future once it finishes execution from the
             writeChunkFutureMap.remove(entryIndex);
@@ -661,6 +657,8 @@ private void handleCommandResult(ContainerCommandRequestProto requestProto, long
         // After concurrent flushes are allowed on the same key, chunk file inconsistencies can happen and
         // that should not crash the pipeline.
         && r.getResult() != ContainerProtos.Result.CHUNK_FILE_INCONSISTENCY) {
+      StorageContainerException sce =
+          new StorageContainerException(r.getMessage(), r.getResult());
       LOG.error(getGroupId() + ": writeChunk writeStateMachineData failed: blockId" +
           write.getBlockID() + " logIndex " + entryIndex + " chunkName " +
           write.getChunkData().getChunkName() + " Error message: " +
@@ -671,7 +669,7 @@ private void handleCommandResult(ContainerCommandRequestProto requestProto, long
       // handling the entry for the write chunk in cache.
       stateMachineHealthy.set(false);
       unhealthyContainers.add(write.getBlockID().getContainerID());
-      raftFuture.complete(r::toByteString);
+      raftFuture.completeExceptionally(sce);
     } else {
       metrics.incNumBytesWrittenCount(
           requestProto.getWriteChunk().getChunkData().getLen());

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java

@@ -22,7 +22,6 @@
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
@@ -58,7 +57,6 @@
 import org.apache.ratis.server.RaftServer;
 import org.apache.ratis.statemachine.TransactionContext;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
-import org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -122,8 +120,7 @@ public void shutdown() {
 
   @ParameterizedTest
   @ValueSource(booleans = {true, false})
-  public void testWriteFailure(boolean failWithException)
-      throws ExecutionException, InterruptedException, InvalidProtocolBufferException {
+  public void testWriteFailure(boolean failWithException) throws ExecutionException, InterruptedException {
     RaftProtos.LogEntryProto entry = mock(RaftProtos.LogEntryProto.class);
     when(entry.getTerm()).thenReturn(1L);
     when(entry.getIndex()).thenReturn(1L);
@@ -137,28 +134,23 @@ public void testWriteFailure(boolean failWithException)
     setUpMockDispatcherReturn(failWithException);
     setUpMockRequestProtoReturn(context, 1, 1);
 
-    Message message = stateMachine.write(entry, trx).get();
+    ThrowableCatcher catcher = new ThrowableCatcher();
+
+    stateMachine.write(entry, trx).exceptionally(catcher.asSetter()).get();
     verify(dispatcher, times(1)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class),
         any(DispatcherContext.class));
     reset(dispatcher);
-    ContainerProtos.ContainerCommandResponseProto containerCommandResponseProto =
-        ContainerProtos.ContainerCommandResponseProto.parseFrom(message.getContent());
-    if (failWithException) {
-      assertEquals("Failed to write chunk data", containerCommandResponseProto.getMessage());
-      assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, containerCommandResponseProto.getResult());
-    } else {
-      // If dispatcher returned failure response, the state machine should propagate the same failure.
-      assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, containerCommandResponseProto.getResult());
-    }
+    assertNotNull(catcher.getReceived());
+    assertResults(failWithException, catcher.getCaught());
 
     // Writing data to another container(containerId 2) should also fail.
     setUpMockRequestProtoReturn(context, 2, 1);
-    message = stateMachine.write(entryNext, trx).get();
+    stateMachine.write(entryNext, trx).exceptionally(catcher.asSetter()).get();
     verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class),
         any(DispatcherContext.class));
-    containerCommandResponseProto
-        = ContainerProtos.ContainerCommandResponseProto.parseFrom(message.getContent());
-    assertTrue(containerCommandResponseProto.getMessage().contains("failed, stopping all writes to container"));
+    assertInstanceOf(StorageContainerException.class, catcher.getReceived());
+    StorageContainerException sce = (StorageContainerException) catcher.getReceived();
+    assertEquals(ContainerProtos.Result.CONTAINER_UNHEALTHY, sce.getResult());
   }
 
   @ParameterizedTest
@@ -232,17 +224,20 @@ public void testWriteTimout() throws Exception {
     }).when(dispatcher).dispatch(any(), any());
 
     setUpMockRequestProtoReturn(context, 1, 1);
+    ThrowableCatcher catcher = new ThrowableCatcher();
 
     CompletableFuture<Message> firstWrite = stateMachine.write(entry, trx);
     Thread.sleep(2000);
     CompletableFuture<Message> secondWrite = stateMachine.write(entryNext, trx);
-    ContainerProtos.ContainerCommandResponseProto containerCommandResponseProto
-        = ContainerProtos.ContainerCommandResponseProto.parseFrom(firstWrite.get().getContent());
-    assertEquals("Failed to write chunk data", containerCommandResponseProto.getMessage());
+    firstWrite.exceptionally(catcher.asSetter()).get();
+    assertNotNull(catcher.getCaught());
+    assertInstanceOf(InterruptedException.class, catcher.getReceived());
 
-    containerCommandResponseProto
-        = ContainerProtos.ContainerCommandResponseProto.parseFrom(secondWrite.get().getContent());
-    assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, containerCommandResponseProto.getResult());
+    secondWrite.exceptionally(catcher.asSetter()).get();
+    assertNotNull(catcher.getReceived());
+    assertInstanceOf(StorageContainerException.class, catcher.getReceived());
+    StorageContainerException sce = (StorageContainerException) catcher.getReceived();
+    assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult());
   }
 
   private void setUpMockDispatcherReturn(boolean failWithException) {

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineFailures.java

@@ -53,7 +53,6 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
@@ -88,21 +87,17 @@
 import org.apache.hadoop.ozone.container.TestHelper;
 import org.apache.hadoop.ozone.container.common.impl.ContainerData;
 import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml;
-import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
 import org.apache.hadoop.ozone.container.common.impl.HddsDispatcher;
 import org.apache.hadoop.ozone.container.common.interfaces.Container;
 import org.apache.hadoop.ozone.container.common.transport.server.ratis.ContainerStateMachine;
 import org.apache.hadoop.ozone.container.common.transport.server.ratis.XceiverServerRatis;
-import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
-import org.apache.hadoop.ozone.container.common.volume.VolumeUsage;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
 import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
 import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand;
 import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
-import org.apache.hadoop.util.Time;
 import org.apache.ozone.test.GenericTestUtils;
 import org.apache.ozone.test.LambdaTestUtils;
 import org.apache.ozone.test.tag.Flaky;
@@ -140,7 +135,7 @@ public static void init() throws Exception {
         TimeUnit.MILLISECONDS);
     conf.setTimeDuration(HDDS_COMMAND_STATUS_REPORT_INTERVAL, 200,
         TimeUnit.MILLISECONDS);
-    conf.setTimeDuration(HDDS_PIPELINE_REPORT_INTERVAL, 2000,
+    conf.setTimeDuration(HDDS_PIPELINE_REPORT_INTERVAL, 200,
         TimeUnit.MILLISECONDS);
     conf.setTimeDuration(HDDS_HEARTBEAT_INTERVAL, 200, TimeUnit.MILLISECONDS);
     conf.setTimeDuration(OZONE_SCM_STALENODE_INTERVAL, 30, TimeUnit.SECONDS);
@@ -770,14 +765,9 @@ void testContainerStateMachineSingleFailureRetry()
       assertEquals(1, locationInfoList.size());
 
       OmKeyLocationInfo omKeyLocationInfo = locationInfoList.get(0);
-      ContainerSet containerSet = cluster.getHddsDatanode(omKeyLocationInfo.getPipeline().getLeaderNode())
-          .getDatanodeStateMachine().getContainer().getContainerSet();
 
       induceFollowerFailure(omKeyLocationInfo, 2);
       key.flush();
-      // wait for container close for failure in flush for both followers applyTransaction failure
-      GenericTestUtils.waitFor(() -> containerSet.getContainer(omKeyLocationInfo.getContainerID()).getContainerData()
-              .getState().equals(ContainerProtos.ContainerDataProto.State.CLOSED), 100, 30000);
       key.write("ratis".getBytes(UTF_8));
       key.flush();
     }
@@ -817,59 +807,6 @@ void testContainerStateMachineDualFailureRetry()
     validateData("ratis1", 2, "ratisratisratisratis");
   }
 
-  @Test
-  void testContainerStateMachineAllNodeFailure()
-      throws Exception {
-    // mark all dn volume as full to induce failure
-    List<Pair<VolumeUsage, Long>> increasedVolumeSpace = new ArrayList<>();
-    cluster.getHddsDatanodes().forEach(
-        dn -> {
-          List<StorageVolume> volumesList = dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
-          volumesList.forEach(sv -> {
-            final VolumeUsage volumeUsage = sv.getVolumeUsage();
-            if (volumeUsage != null) {
-              final long available = sv.getCurrentUsage().getAvailable();
-              increasedVolumeSpace.add(Pair.of(volumeUsage, available));
-              volumeUsage.incrementUsedSpace(available);
-            }
-          });
-        }
-    );
-
-    long startTime = Time.monotonicNow();
-    ReplicationConfig replicationConfig = ReplicationConfig.fromTypeAndFactor(ReplicationType.RATIS,
-        ReplicationFactor.THREE);
-    try (OzoneOutputStream key = objectStore.getVolume(volumeName).getBucket(bucketName).createKey(
-        "testkey1", 1024, replicationConfig, new HashMap<>())) {
-
-      key.write("ratis".getBytes(UTF_8));
-      key.flush();
-      fail();
-    } catch (IOException ex) {
-      assertThat(ex.getMessage()).contains("Retry request failed. retries get failed due to exceeded" +
-          " maximum allowed retries number: 5");
-    } finally {
-      increasedVolumeSpace.forEach(e -> e.getLeft().decrementUsedSpace(e.getRight()));
-      // test execution is less than 2 sec but to be safe putting 30 sec as without fix, taking more than 60 sec
-      assertThat(Time.monotonicNow() - startTime)
-          .describedAs("Operation took longer than expected")
-          .isLessThan(30000);
-    }
-
-    // previous pipeline gets closed due to disk full failure, so created a new pipeline and write should succeed,
-    // and this ensures later test case can pass (should not fail due to pipeline unavailability as timeout is 200ms
-    // for pipeline creation which can fail in testcase later on)
-    Pipeline pipeline = cluster.getStorageContainerManager().getPipelineManager().createPipeline(replicationConfig);
-    cluster.getStorageContainerManager().getPipelineManager().waitPipelineReady(pipeline.getId(), 60000);
-
-    try (OzoneOutputStream key = objectStore.getVolume(volumeName).getBucket(bucketName).createKey(
-        "testkey2", 1024, replicationConfig, new HashMap<>())) {
-
-      key.write("ratis".getBytes(UTF_8));
-      key.flush();
-    }
-  }
-
   private void induceFollowerFailure(OmKeyLocationInfo omKeyLocationInfo,
                                      int failureCount) {
     DatanodeID leader = omKeyLocationInfo.getPipeline().getLeaderId();