Skip to content

Commit

Permalink
test: fix flaky HttpJsonDirectServerStreamingCallableTest.testOnRespo…
Browse files Browse the repository at this point in the history 
…nseError (#2444)

* increase timeout from 2s to 30s to reduce the chance of
DEADLINE_EXCEEDED before the NOT_FOUND message is received
* keep separate low timeout for the testDeadlineExceededServerStreaming
test
* remove mockService reset to speedup tests, as it's no longer necessary

Fixes: #1842.
  • Loading branch information
@meltsufin
meltsufin authored Feb 7, 2024
1 parent b0a57b7 commit 564802b
Showing 1 changed file with 50 additions and 51 deletions.
101 changes: 50 additions & 51 deletions .../src/test/java/com/google/api/gax/httpjson/HttpJsonDirectServerStreamingCallableTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,63 +72,65 @@

@RunWith(JUnit4.class)
public class HttpJsonDirectServerStreamingCallableTest {
private static final ApiMethodDescriptor<Color, Money> METHOD_SERVER_STREAMING_RECOGNIZE =
ApiMethodDescriptor.<Color, Money>newBuilder()
.setFullMethodName("google.cloud.v1.Fake/ServerStreamingRecognize")
.setHttpMethod("POST")
.setRequestFormatter(
ProtoMessageRequestFormatter.<Color>newBuilder()
.setPath(
"/fake/v1/recognize/{blue}",
request -> {
Map<String, String> fields = new HashMap<>();
ProtoRestSerializer<Field> serializer = ProtoRestSerializer.create();
serializer.putPathParam(fields, "blue", request.getBlue());
return fields;
})
.setQueryParamsExtractor(
request -> {
Map<String, List<String>> fields = new HashMap<>();
ProtoRestSerializer<Field> serializer = ProtoRestSerializer.create();
serializer.putQueryParam(fields, "red", request.getRed());
return fields;
})
.setRequestBodyExtractor(
request ->
ProtoRestSerializer.create()
.toBody(
"*", request.toBuilder().clearBlue().clearRed().build(), false))
.build())
.setResponseParser(
ProtoMessageResponseParser.<Money>newBuilder()
.setDefaultInstance(Money.getDefaultInstance())
.build())
.setType(MethodType.SERVER_STREAMING)
.build();

private MockHttpService mockService;

private static final Color DEFAULT_REQUEST = Color.newBuilder().setRed(0.5f).build();
private static final Color ASYNC_REQUEST = DEFAULT_REQUEST.toBuilder().setGreen(1000).build();
private static final Color ERROR_REQUEST = Color.newBuilder().setRed(-1).build();
private static final Money DEFAULT_RESPONSE =
Money.newBuilder().setCurrencyCode("USD").setUnits(127).build();
private static final Money DEFAULTER_RESPONSE =
Money.newBuilder().setCurrencyCode("UAH").setUnits(255).build();
private static final int AWAIT_TERMINATION_SECONDS = 10;

private ServerStreamingCallSettings<Color, Money> streamingCallSettings;
private ServerStreamingCallable<Color, Money> streamingCallable;

private ManagedHttpJsonChannel channel;
private ClientContext clientContext;
private ExecutorService executorService;
private MockHttpService mockService;
ApiMethodDescriptor<Color, Money> methodServerStreamingRecognize;

@Before
public void initialize() throws IOException {
mockService =
initialize(Duration.ofSeconds(30));
}

public void initialize(Duration timeout) throws IOException {
this.methodServerStreamingRecognize =
ApiMethodDescriptor.<Color, Money>newBuilder()
.setFullMethodName("google.cloud.v1.Fake/ServerStreamingRecognize")
.setHttpMethod("POST")
.setRequestFormatter(
ProtoMessageRequestFormatter.<Color>newBuilder()
.setPath(
"/fake/v1/recognize/{blue}",
request -> {
Map<String, String> fields = new HashMap<>();
ProtoRestSerializer<Field> serializer = ProtoRestSerializer.create();
serializer.putPathParam(fields, "blue", request.getBlue());
return fields;
})
.setQueryParamsExtractor(
request -> {
Map<String, List<String>> fields = new HashMap<>();
ProtoRestSerializer<Field> serializer = ProtoRestSerializer.create();
serializer.putQueryParam(fields, "red", request.getRed());
return fields;
})
.setRequestBodyExtractor(
request ->
ProtoRestSerializer.create()
.toBody(
"*", request.toBuilder().clearBlue().clearRed().build(), false))
.build())
.setResponseParser(
ProtoMessageResponseParser.<Money>newBuilder()
.setDefaultInstance(Money.getDefaultInstance())
.build())
.setType(MethodType.SERVER_STREAMING)
.build();
this.mockService =
new MockHttpService(
Collections.singletonList(METHOD_SERVER_STREAMING_RECOGNIZE), "google.com:443");
Collections.singletonList(methodServerStreamingRecognize), "google.com:443");
executorService = Executors.newFixedThreadPool(2);
channel =
new ManagedHttpJsonInterceptorChannel(
Expand All @@ -148,28 +150,22 @@ public void initialize() throws IOException {
.setTransportChannel(HttpJsonTransportChannel.create(channel))
.setDefaultCallContext(
HttpJsonCallContext.of(channel, HttpJsonCallOptions.DEFAULT)
.withTimeout(Duration.ofSeconds(3))
.withTimeout(timeout)
.withEndpointContext(endpointContext))
.build();

streamingCallSettings = ServerStreamingCallSettings.<Color, Money>newBuilder().build();
streamingCallable =
HttpJsonCallableFactory.createServerStreamingCallable(
HttpJsonCallSettings.create(METHOD_SERVER_STREAMING_RECOGNIZE),
HttpJsonCallSettings.create(methodServerStreamingRecognize),
streamingCallSettings,
clientContext);

mockService.reset();
}

@After
public void destroy() throws InterruptedException {
executorService.shutdown();
channel.shutdown();

executorService.awaitTermination(AWAIT_TERMINATION_SECONDS, TimeUnit.SECONDS);
channel.awaitTermination(AWAIT_TERMINATION_SECONDS, TimeUnit.SECONDS);
mockService.reset();
}

@Test
Expand All @@ -178,7 +174,7 @@ public void testBadContext() {
// Create a local callable with a bad context
ServerStreamingCallable<Color, Money> streamingCallable =
HttpJsonCallableFactory.createServerStreamingCallable(
HttpJsonCallSettings.create(METHOD_SERVER_STREAMING_RECOGNIZE),
HttpJsonCallSettings.create(this.methodServerStreamingRecognize),
streamingCallSettings,
clientContext
.toBuilder()
Expand Down Expand Up @@ -337,9 +333,12 @@ public void testBlockingServerStreaming() {

// This test ensures that the server-side streaming does not exceed the timeout value
@Test
public void testDeadlineExceededServerStreaming() throws InterruptedException {
public void testDeadlineExceededServerStreaming() throws InterruptedException, IOException {
// set a low timeout to trigger deadline-exceeded sooner
initialize(Duration.ofSeconds(1));

mockService.addResponse(
new Money[] {DEFAULT_RESPONSE, DEFAULTER_RESPONSE}, java.time.Duration.ofSeconds(5));
new Money[] {DEFAULT_RESPONSE, DEFAULTER_RESPONSE}, java.time.Duration.ofSeconds(30));
Color request = Color.newBuilder().setRed(0.5f).build();
CountDownLatch latch = new CountDownLatch(1);
MoneyObserver moneyObserver = new MoneyObserver(false, latch);
Expand All @@ -349,7 +348,7 @@ public void testDeadlineExceededServerStreaming() throws InterruptedException {
moneyObserver.controller.request(2);
// Set the latch's await time to above the context's timeout value to ensure that
// the latch has been released.
Truth.assertThat(latch.await(5000, TimeUnit.MILLISECONDS)).isTrue();
Truth.assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();

Truth.assertThat(moneyObserver.error).isInstanceOf(DeadlineExceededException.class);
Truth.assertThat(moneyObserver.error).hasMessageThat().isEqualTo("Deadline exceeded");
Expand Down

0 comments on commit 564802b

Please sign in to comment.