Skip to content
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

core: updates the backoff range as per the A6 redefinition #11858

Merged
merged 5 commits into from
Feb 5, 2025
Merged

core: updates the backoff range as per the A6 redefinition #11858

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f3ed6ef
core: updates the backoff range being used from [0, 1] to [0.8, 1.2] …
AgraVator Jan 28, 2025
29824e3
adds a flag for experimental jitter
AgraVator Jan 30, 2025
c2f5c1c
xds: Allow FaultFilter's interceptor to be reused
ejona86 Jan 29, 2025
c3244f6
netty: Removed 4096 min buffer size (#11856)
AgraVator Jan 30, 2025
dad68ff
turns the flag in a var for better efficiency
AgraVator Feb 5, 2025
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
11 changes: 10 additions & 1 deletion core/src/main/java/io/grpc/internal/RetriableStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -846,6 +846,15 @@ public void run() {
}
}

private static final boolean isExperimentalRetryJitterEnabled = GrpcUtil
.getFlag("GRPC_EXPERIMENTAL_XDS_RLS_LB", true);

public static long intervalWithJitter(long intervalNanos) {
double inverseJitterFactor = isExperimentalRetryJitterEnabled
? 0.8 * random.nextDouble() + 0.4 : random.nextDouble();
return (long) (intervalNanos * inverseJitterFactor);
}

private static final class SavedCloseMasterListenerReason {
private final Status status;
private final RpcProgress progress;
Expand Down Expand Up @@ -1066,7 +1075,7 @@ private RetryPlan makeRetryDecision(Status status, Metadata trailer) {
if (pushbackMillis == null) {
if (isRetryableStatusCode) {
shouldRetry = true;
backoffNanos = (long) (nextBackoffIntervalNanos * random.nextDouble());
backoffNanos = intervalWithJitter(nextBackoffIntervalNanos);
nextBackoffIntervalNanos = Math.min(
(long) (nextBackoffIntervalNanos * retryPolicy.backoffMultiplier),
retryPolicy.maxBackoffNanos);
Expand Down
4 changes: 2 additions & 2 deletions core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3875,7 +3875,7 @@ public double nextDouble() {
Status.UNAVAILABLE, PROCESSED, new Metadata());

// in backoff
timer.forwardTime(5, TimeUnit.SECONDS);
timer.forwardTime(6, TimeUnit.SECONDS);
assertThat(timer.getPendingTasks()).hasSize(1);
verify(mockStream2, never()).start(any(ClientStreamListener.class));

Expand All @@ -3894,7 +3894,7 @@ public double nextDouble() {
assertEquals("Channel shutdown invoked", statusCaptor.getValue().getDescription());

// backoff ends
timer.forwardTime(5, TimeUnit.SECONDS);
timer.forwardTime(6, TimeUnit.SECONDS);
assertThat(timer.getPendingTasks()).isEmpty();
verify(mockStream2).start(streamListenerCaptor.capture());
verify(mockLoadBalancer, never()).shutdown();
Expand Down
79 changes: 38 additions & 41 deletions core/src/test/java/io/grpc/internal/RetriableStreamTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,17 @@ public double nextDouble() {
private final ChannelBufferMeter channelBufferUsed = new ChannelBufferMeter();
private final FakeClock fakeClock = new FakeClock();

private static long calculateBackoffWithRetries(int retryCount) {
// Calculate the exponential backoff delay with jitter
double exponent = retryCount > 0 ? Math.pow(BACKOFF_MULTIPLIER, retryCount) : 1;
long delay = (long) (INITIAL_BACKOFF_IN_SECONDS * exponent);
return RetriableStream.intervalWithJitter(delay);
Copy link
Member

@shivaspeaks shivaspeaks Jan 30, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

This is not testing when isExperimentalRetryJitterEnabled is toggled.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yes, had discussed it with @kannanjgithub, we can add it if it is absolutely needed, but the default behavior is covered and the other one is to be deprecated after a few releases.
P.S. I think you meant the testing is not covered for "false" or the old behavior.

Copy link
Contributor

Choose a reason for hiding this comment

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

In this case it is pretty trivial to see that setting the flag to false couldn't cause any problems, so it is fine to skip testing that case.

If it wasn't so obvious, but still simple, you'd want to create a single test case where you:

  1. saved the flag's System.property value (could be null)
  2. set it to the non-default value (in this case "false")
  3. Do the test logic
  4. restore the System.property (for null do an unset)
    Then when the flag is removed the test case can be deleted (would want to add a TODO on the test case indicating that it should be removed that includes the flag name).

If it was something complicated that affected a lot of behavior then you would want to parameterize the test so that all (or at least most) of the test cases would run both ways. You can do a search for @RunWith(Parameterized.class) to see a number of examples in our existing tests. Note that to skip specific tests with some options you can use org.junit.Assume

}

private static long calculateMaxBackoff() {
return RetriableStream.intervalWithJitter(MAX_BACKOFF_IN_SECONDS);
}

private final class RecordedRetriableStream extends RetriableStream<String> {
RecordedRetriableStream(MethodDescriptor<String, ?> method, Metadata headers,
ChannelBufferMeter channelBufferUsed, long perRpcBufferLimit, long channelBufferLimit,
Expand Down Expand Up @@ -307,7 +318,7 @@ public Void answer(InvocationOnMock in) {
retriableStream.sendMessage("msg1 during backoff1");
retriableStream.sendMessage("msg2 during backoff1");

fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM) - 1L, TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0) - 1L, TimeUnit.SECONDS);
inOrder.verifyNoMoreInteractions();
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
Expand Down Expand Up @@ -364,9 +375,7 @@ public Void answer(InvocationOnMock in) {
retriableStream.sendMessage("msg2 during backoff2");
retriableStream.sendMessage("msg3 during backoff2");

fakeClock.forwardTime(
(long) (INITIAL_BACKOFF_IN_SECONDS * BACKOFF_MULTIPLIER * FAKE_RANDOM) - 1L,
TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(1) - 1L, TimeUnit.SECONDS);
inOrder.verifyNoMoreInteractions();
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
Expand Down Expand Up @@ -459,7 +468,7 @@ public void retry_headersRead_cancel() {
sublistenerCaptor1.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
ArgumentCaptor.forClass(ClientStreamListener.class);
Expand Down Expand Up @@ -518,7 +527,7 @@ public void retry_headersRead_closed() {
doReturn(mockStream2).when(retriableStreamRecorder).newSubstream(1);
sublistenerCaptor1.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
ArgumentCaptor.forClass(ClientStreamListener.class);
Expand Down Expand Up @@ -584,7 +593,7 @@ public void retry_cancel_closed() {
doReturn(mockStream2).when(retriableStreamRecorder).newSubstream(1);
sublistenerCaptor1.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
ArgumentCaptor.forClass(ClientStreamListener.class);
Expand Down Expand Up @@ -687,7 +696,7 @@ public void retry_unretriableClosed_cancel() {
doReturn(mockStream2).when(retriableStreamRecorder).newSubstream(1);
sublistenerCaptor1.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
ArgumentCaptor.forClass(ClientStreamListener.class);
Expand Down Expand Up @@ -821,7 +830,7 @@ public boolean isReady() {
// send more requests during backoff
retriableStream.request(789);

fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

inOrder.verify(mockStream2).start(sublistenerCaptor2.get());
inOrder.verify(mockStream2).request(3);
Expand Down Expand Up @@ -875,7 +884,7 @@ public void request(int numMessages) {
doReturn(mockStream2).when(retriableStreamRecorder).newSubstream(1);
sublistenerCaptor1.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

inOrder.verify(mockStream2).start(sublistenerCaptor2.capture());
inOrder.verify(mockStream2).request(3);
Expand Down Expand Up @@ -920,7 +929,7 @@ public void start(ClientStreamListener listener) {
doReturn(mockStream2).when(retriableStreamRecorder).newSubstream(1);
sublistenerCaptor1.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

inOrder.verify(mockStream2).start(sublistenerCaptor2.capture());
inOrder.verify(retriableStreamRecorder).postCommit();
Expand Down Expand Up @@ -1028,7 +1037,7 @@ public boolean isReady() {
retriableStream.request(789);
readiness.add(retriableStream.isReady()); // expected false b/c in backoff

fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

verify(mockStream2).start(any(ClientStreamListener.class));
readiness.add(retriableStream.isReady()); // expected true
Expand Down Expand Up @@ -1110,7 +1119,7 @@ public void addPrevRetryAttemptsToRespHeaders() {
doReturn(mockStream2).when(retriableStreamRecorder).newSubstream(1);
sublistenerCaptor1.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
ArgumentCaptor.forClass(ClientStreamListener.class);
Expand Down Expand Up @@ -1160,13 +1169,12 @@ public void start(ClientStreamListener listener) {
listener1.closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);
assertEquals(1, fakeClock.numPendingTasks());

// send requests during backoff
retriableStream.request(3);
fakeClock.forwardTime(
(long) (INITIAL_BACKOFF_IN_SECONDS * BACKOFF_MULTIPLIER * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(1), TimeUnit.SECONDS);

retriableStream.request(1);
verify(mockStream1, never()).request(anyInt());
Expand Down Expand Up @@ -1207,7 +1215,7 @@ public void start(ClientStreamListener listener) {
// retry
listener1.closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);

verify(mockStream2).start(any(ClientStreamListener.class));
verify(retriableStreamRecorder).postCommit();
Expand Down Expand Up @@ -1260,7 +1268,7 @@ public void perRpcBufferLimitExceededDuringBackoff() {
bufferSizeTracer.outboundWireSize(2);
verify(retriableStreamRecorder, never()).postCommit();

fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);
verify(mockStream2).start(any(ClientStreamListener.class));
verify(mockStream2).isReady();

Expand Down Expand Up @@ -1332,7 +1340,7 @@ public void expBackoff_maxBackoff_maxRetryAttempts() {
sublistenerCaptor1.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM) - 1L, TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0) - 1L, TimeUnit.SECONDS);
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
assertEquals(0, fakeClock.numPendingTasks());
Expand All @@ -1347,9 +1355,7 @@ public void expBackoff_maxBackoff_maxRetryAttempts() {
sublistenerCaptor2.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_2), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(
(long) (INITIAL_BACKOFF_IN_SECONDS * BACKOFF_MULTIPLIER * FAKE_RANDOM) - 1L,
TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(1) - 1L, TimeUnit.SECONDS);
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
assertEquals(0, fakeClock.numPendingTasks());
Expand All @@ -1364,10 +1370,7 @@ public void expBackoff_maxBackoff_maxRetryAttempts() {
sublistenerCaptor3.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(
(long) (INITIAL_BACKOFF_IN_SECONDS * BACKOFF_MULTIPLIER * BACKOFF_MULTIPLIER * FAKE_RANDOM)
- 1L,
TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(2) - 1L, TimeUnit.SECONDS);
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
assertEquals(0, fakeClock.numPendingTasks());
Expand All @@ -1382,7 +1385,7 @@ public void expBackoff_maxBackoff_maxRetryAttempts() {
sublistenerCaptor4.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_2), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime((long) (MAX_BACKOFF_IN_SECONDS * FAKE_RANDOM) - 1L, TimeUnit.SECONDS);
fakeClock.forwardTime(calculateMaxBackoff() - 1L, TimeUnit.SECONDS);
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
assertEquals(0, fakeClock.numPendingTasks());
Expand All @@ -1397,7 +1400,7 @@ public void expBackoff_maxBackoff_maxRetryAttempts() {
sublistenerCaptor5.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_2), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime((long) (MAX_BACKOFF_IN_SECONDS * FAKE_RANDOM) - 1L, TimeUnit.SECONDS);
fakeClock.forwardTime(calculateMaxBackoff() - 1L, TimeUnit.SECONDS);
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
assertEquals(0, fakeClock.numPendingTasks());
Expand Down Expand Up @@ -1480,7 +1483,7 @@ public void pushback() {
sublistenerCaptor3.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM) - 1L, TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0) - 1L, TimeUnit.SECONDS);
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
assertEquals(0, fakeClock.numPendingTasks());
Expand All @@ -1495,9 +1498,7 @@ public void pushback() {
sublistenerCaptor4.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_2), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(
(long) (INITIAL_BACKOFF_IN_SECONDS * BACKOFF_MULTIPLIER * FAKE_RANDOM) - 1L,
TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(1) - 1L, TimeUnit.SECONDS);
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
assertEquals(0, fakeClock.numPendingTasks());
Expand All @@ -1512,10 +1513,7 @@ public void pushback() {
sublistenerCaptor5.getValue().closed(
Status.fromCode(RETRIABLE_STATUS_CODE_2), PROCESSED, new Metadata());
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(
(long) (INITIAL_BACKOFF_IN_SECONDS * BACKOFF_MULTIPLIER * BACKOFF_MULTIPLIER * FAKE_RANDOM)
- 1L,
TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(2) - 1L, TimeUnit.SECONDS);
assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(1L, TimeUnit.SECONDS);
assertEquals(0, fakeClock.numPendingTasks());
Expand Down Expand Up @@ -1804,7 +1802,7 @@ public void transparentRetry_onlyOnceOnRefused() {
.closed(Status.fromCode(RETRIABLE_STATUS_CODE_1), REFUSED, new Metadata());

assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);
inOrder.verify(retriableStreamRecorder).newSubstream(1);
ArgumentCaptor<ClientStreamListener> sublistenerCaptor3 =
ArgumentCaptor.forClass(ClientStreamListener.class);
Expand Down Expand Up @@ -1907,7 +1905,7 @@ public void normalRetry_thenNoTransparentRetry_butNormalRetry() {
.closed(Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());

assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);
inOrder.verify(retriableStreamRecorder).newSubstream(1);
ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
ArgumentCaptor.forClass(ClientStreamListener.class);
Expand All @@ -1923,8 +1921,7 @@ public void normalRetry_thenNoTransparentRetry_butNormalRetry() {
.closed(Status.fromCode(RETRIABLE_STATUS_CODE_1), REFUSED, new Metadata());

assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime(
(long) (INITIAL_BACKOFF_IN_SECONDS * BACKOFF_MULTIPLIER * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(1), TimeUnit.SECONDS);
inOrder.verify(retriableStreamRecorder).newSubstream(2);
ArgumentCaptor<ClientStreamListener> sublistenerCaptor3 =
ArgumentCaptor.forClass(ClientStreamListener.class);
Expand Down Expand Up @@ -1960,7 +1957,7 @@ public void normalRetry_thenNoTransparentRetry_andNoMoreRetry() {
.closed(Status.fromCode(RETRIABLE_STATUS_CODE_1), PROCESSED, new Metadata());

assertEquals(1, fakeClock.numPendingTasks());
fakeClock.forwardTime((long) (INITIAL_BACKOFF_IN_SECONDS * FAKE_RANDOM), TimeUnit.SECONDS);
fakeClock.forwardTime(calculateBackoffWithRetries(0), TimeUnit.SECONDS);
inOrder.verify(retriableStreamRecorder).newSubstream(1);
ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
ArgumentCaptor.forClass(ClientStreamListener.class);
Expand Down
10 changes: 5 additions & 5 deletions interop-testing/src/test/java/io/grpc/testing/integration/RetryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -303,7 +303,7 @@ public void retryUntilBufferLimitExceeded() throws Exception {
serverCall.close(
Status.UNAVAILABLE.withDescription("original attempt failed"),
new Metadata());
elapseBackoff(10, SECONDS);
elapseBackoff(12, SECONDS);
// 2nd attempt received
serverCall = serverCalls.poll(5, SECONDS);
serverCall.request(2);
Expand Down Expand Up @@ -348,7 +348,7 @@ public void statsRecorded() throws Exception {
Status.UNAVAILABLE.withDescription("original attempt failed"),
new Metadata());
assertRpcStatusRecorded(Status.Code.UNAVAILABLE, 1000, 1);
elapseBackoff(10, SECONDS);
elapseBackoff(12, SECONDS);
assertRpcStartedRecorded();
assertOutboundMessageRecorded();
serverCall = serverCalls.poll(5, SECONDS);
Expand All @@ -366,7 +366,7 @@ public void statsRecorded() throws Exception {
call.request(1);
assertInboundMessageRecorded();
assertInboundWireSizeRecorded(1);
assertRpcStatusRecorded(Status.Code.OK, 12000, 2);
assertRpcStatusRecorded(Status.Code.OK, 14000, 2);
assertRetryStatsRecorded(1, 0, 0);
}

Expand Down Expand Up @@ -418,7 +418,7 @@ public void streamClosed(Status status) {
Status.UNAVAILABLE.withDescription("original attempt failed"),
new Metadata());
assertRpcStatusRecorded(Code.UNAVAILABLE, 5000, 1);
elapseBackoff(10, SECONDS);
elapseBackoff(12, SECONDS);
assertRpcStartedRecorded();
assertOutboundMessageRecorded();
serverCall = serverCalls.poll(5, SECONDS);
Expand All @@ -431,7 +431,7 @@ public void streamClosed(Status status) {
streamClosedLatch.countDown();
// The call listener is closed.
verify(mockCallListener, timeout(5000)).onClose(any(Status.class), any(Metadata.class));
assertRpcStatusRecorded(Code.CANCELLED, 17_000, 1);
assertRpcStatusRecorded(Code.CANCELLED, 19_000, 1);
assertRetryStatsRecorded(1, 0, 0);
}

Expand Down
5 changes: 1 addition & 4 deletions netty/src/main/java/io/grpc/netty/NettyWritableBufferAllocator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,9 +33,6 @@
*/
class NettyWritableBufferAllocator implements WritableBufferAllocator {

// Use 4k as our minimum buffer size.
private static final int MIN_BUFFER = 4 * 1024;

// Set the maximum buffer size to 1MB.
private static final int MAX_BUFFER = 1024 * 1024;

Expand All @@ -47,7 +44,7 @@ class NettyWritableBufferAllocator implements WritableBufferAllocator {

@Override
public WritableBuffer allocate(int capacityHint) {
capacityHint = Math.min(MAX_BUFFER, Math.max(MIN_BUFFER, capacityHint));
capacityHint = Math.min(MAX_BUFFER, capacityHint);
return new NettyWritableBuffer(allocator.buffer(capacityHint, capacityHint));
}
}
Loading