Skip to content

HDDS-10427. Retry read wait based on policy. #6292

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
ChenSammi merged 2 commits into from
Mar 4, 2024
+21 −1
Merged

HDDS-10427. Retry read wait based on policy. #6292

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
757572c
HDDS-10427. Retry read wait based on policy.
ashishkr200 Feb 28, 2024
03cede6
Fix review comment
ashishkr200 Feb 29, 2024
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
15 changes: 14 additions & 1 deletion hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -574,7 +574,20 @@ private boolean shouldRetryRead(IOException cause) throws IOException {
} catch (Exception e) {
throw new IOException(e);
}
return retryAction.action == RetryPolicy.RetryAction.RetryDecision.RETRY;
if (retryAction.action == RetryPolicy.RetryAction.RetryDecision.RETRY) {
Copy link
Contributor

@ivandika3 ivandika3 Feb 29, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just a thought: It might be nice if we can have a general RetryAction handler so that we don't need to duplicate the sleeping logic (similar codes is in AbstractDataStreamOutput#shouldRetry, GrpcOmTranport#shouldRetry, and KeyOutputStream#handleRetry)

Copy link
Contributor Author

@ashishkumar50 ashishkumar50 Feb 29, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes sounds good may be we can refactor in other Jira.

if (retryAction.delayMillis > 0) {
try {
LOG.debug("Retry read after {}ms", retryAction.delayMillis);
Thread.sleep(retryAction.delayMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
String msg = "Interrupted: action=" + retryAction.action + ", retry policy=" + retryPolicy;
throw new IOException(msg, e);
}
}
return true;
}
return false;
}

private void handleReadError(IOException cause) throws IOException {
Expand Down
7 changes: 7 additions & 0 deletions ...op-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.hadoop.ozone.common.Checksum;

import org.apache.hadoop.ozone.common.OzoneChecksumException;
import org.apache.ozone.test.GenericTestUtils;
import org.apache.ratis.thirdparty.io.grpc.Status;
import org.apache.ratis.thirdparty.io.grpc.StatusException;
import org.junit.jupiter.api.BeforeEach;
Expand All @@ -42,6 +43,7 @@
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.stubbing.OngoingStubbing;
import org.slf4j.event.Level;

import java.io.EOFException;
import java.io.IOException;
Expand All @@ -58,6 +60,7 @@

import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_NOT_FOUND;
import static org.apache.hadoop.hdds.scm.storage.TestChunkInputStream.generateRandomData;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
Expand Down Expand Up @@ -257,6 +260,9 @@ public void testSeekAndRead() throws Exception {

@Test
public void testRefreshPipelineFunction() throws Exception {
GenericTestUtils.LogCapturer logCapturer = GenericTestUtils.LogCapturer
.captureLogs(BlockInputStream.LOG);
GenericTestUtils.setLogLevel(BlockInputStream.LOG, Level.DEBUG);
BlockID blockID = new BlockID(new ContainerBlockID(1, 1));
AtomicBoolean isRefreshed = new AtomicBoolean();
createChunkList(5);
Expand All @@ -269,6 +275,7 @@ public void testRefreshPipelineFunction() throws Exception {
seekAndVerify(50);
byte[] b = new byte[200];
blockInputStreamWithRetry.read(b, 0, 200);
assertThat(logCapturer.getOutput()).contains("Retry read after");
Copy link
Contributor

@jojochuang jojochuang Feb 29, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Verifying debug log message is going to be brittle. But writing a test to verify time is always tricky. What I usually do is to introduce a timer class that wraps around Thread.sleep(), and we can verify the time does advance (artifically). That's going to introduce big code churn with little benefit so I'm okay with what we have now.

assertTrue(isRefreshed.get());
}
}
Expand Down