Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sumitagrawl merged 1 commit into from
Apr 27, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 6 additions & 8 deletions ...rg/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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);
Expand All @@ -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: " +
Expand All @@ -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());
Expand Down
43 changes: 19 additions & 24 deletions ...pache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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
Expand Down Expand Up @@ -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) {
Expand Down
65 changes: 1 addition & 64 deletions ...t/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineFailures.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();
}
Expand Down Expand Up @@ -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();
Expand Down