Skip to content

Commit c17f9b4

Browse files
kannanjgithubkannanjgithub
committed
xds: avoid IllegalStateException on sendMessage/halfClose after internal cancellation
When RawMessageClientInterceptor or ExternalProcessorClientInterceptor internally cancels the downstream call (e.g., on an IOException while reading the request stream in sendMessage), treat subsequent sendMessage and halfClose calls from the application as no-ops rather than throwing IllegalStateException, while preserving IllegalStateException after explicit application-initiated call.cancel(). TAG=agy CONV=56c87717-243d-4f12-af9e-c532e509892a
1 parent 15d64a4 commit c17f9b4

4 files changed

Lines changed: 202 additions & 24 deletions

File tree

‎xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
package io.grpc.xds;
1818

1919
import static com.google.common.base.Preconditions.checkNotNull;
20+
import static com.google.common.base.Preconditions.checkState;
2021
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.applyHeaderMutations;
2122
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.collectAttributes;
2223
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.markDataPlaneCallClosed;
@@ -366,6 +367,8 @@ private static class DataPlaneClientCall<ReqT, RespT>
366367
final AtomicBoolean isProcessingTrailers = new AtomicBoolean(false);
367368
final AtomicBoolean pendingHalfClose = new AtomicBoolean(false);
368369
final AtomicBoolean bodyMessageSentToExtProc = new AtomicBoolean(false);
370+
final AtomicBoolean aborted = new AtomicBoolean(false);
371+
final AtomicBoolean cancelCalled = new AtomicBoolean(false);
369372

370373
protected DataPlaneClientCall(
371374
DataPlaneDelayedCall<ReqT, RespT> delayedCall,
@@ -994,6 +997,10 @@ public void request(int numMessages) {
994997
@SuppressWarnings("unchecked")
995998
@Override
996999
public void sendMessage(ReqT message) {
1000+
if (aborted.get()) {
1001+
return;
1002+
}
1003+
checkState(!cancelCalled.get(), "call was cancelled");
9971004
if (requestSideClosed.get()) {
9981005
// External processor already closed the request stream. Discard further messages.
9991006
return;
@@ -1080,6 +1087,9 @@ private void drainPendingRequestBodyMessages() {
10801087
}
10811088

10821089
private void proceedWithHalfClose() {
1090+
if (aborted.get()) {
1091+
return;
1092+
}
10831093
if (clientHalfCloseStartNanos > 0) {
10841094
long durationNanos = System.nanoTime() - clientHalfCloseStartNanos;
10851095
recordDuration(clientHalfCloseDuration, durationNanos);
@@ -1090,6 +1100,10 @@ private void proceedWithHalfClose() {
10901100

10911101
@Override
10921102
public void halfClose() {
1103+
if (aborted.get()) {
1104+
return;
1105+
}
1106+
checkState(!cancelCalled.get(), "call was cancelled");
10931107
if (appHalfClosed.compareAndSet(false, true)) {
10941108
clientHalfCloseStartNanos = System.nanoTime();
10951109
}
@@ -1159,11 +1173,16 @@ public void halfClose() {
11591173
}
11601174

11611175
void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {
1176+
if (!cancelCalled.get()) {
1177+
aborted.set(true);
1178+
}
11621179
delayedCall.cancel(message, cause);
11631180
}
11641181

11651182
@Override
11661183
public void cancel(@Nullable String message, @Nullable Throwable cause) {
1184+
cancelCalled.set(true);
1185+
aborted.set(false);
11671186
synchronized (streamLock) {
11681187
if (markExtProcStreamFailed(extProcStreamState)) {
11691188
if (extProcClientCallRequestObserver != null) {

‎xds/src/main/java/io/grpc/xds/XdsNameResolver.java‎

Lines changed: 40 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1283,35 +1283,46 @@ public void onReady() {
12831283
};
12841284
}
12851285

1286-
private static <ReqT> void sendSerializedMessage(
1287-
ClientCall<InputStream, ?> rawCall,
1288-
MethodDescriptor<ReqT, ?> method,
1289-
ReqT message) {
1290-
ByteString byteString;
1291-
try {
1292-
InputStream stream = method.getRequestMarshaller().stream(message);
1286+
private abstract static class PartialForwardingClientCall<ReqT, RespT>
1287+
extends ClientCall<ReqT, RespT> {
1288+
private boolean aborted;
1289+
private boolean cancelCalled;
1290+
1291+
protected abstract ClientCall<?, ?> delegate();
1292+
1293+
protected void sendSerializedMessage(
1294+
ClientCall<InputStream, ?> rawCall,
1295+
MethodDescriptor<ReqT, ?> method,
1296+
ReqT message) {
1297+
if (aborted) {
1298+
return;
1299+
}
1300+
if (cancelCalled) {
1301+
rawCall.sendMessage(null);
1302+
return;
1303+
}
1304+
ByteString byteString;
12931305
try {
1294-
byteString = ExternalProcessorUtil.outboundStreamToByteString(stream);
1295-
} finally {
1306+
InputStream stream = method.getRequestMarshaller().stream(message);
12961307
try {
1297-
stream.close();
1298-
} catch (IOException ignored) {
1299-
// ignore
1308+
byteString = ExternalProcessorUtil.outboundStreamToByteString(stream);
1309+
} finally {
1310+
try {
1311+
stream.close();
1312+
} catch (IOException ignored) {
1313+
// ignore
1314+
}
13001315
}
1316+
} catch (IOException | RuntimeException e) {
1317+
aborted = true;
1318+
rawCall.cancel("Failed to read message for raw message interceptor", e);
1319+
return;
1320+
} catch (Error e) {
1321+
rawCall.cancel("Client sendMessage() failed with Error", e);
1322+
throw e;
13011323
}
1302-
} catch (IOException | RuntimeException e) {
1303-
rawCall.cancel("Failed to read message for raw message interceptor", e);
1304-
return;
1305-
} catch (Error e) {
1306-
rawCall.cancel("Client sendMessage() failed with Error", e);
1307-
throw e;
1324+
rawCall.sendMessage(new KnownLengthInputStream(byteString));
13081325
}
1309-
rawCall.sendMessage(new KnownLengthInputStream(byteString));
1310-
}
1311-
1312-
private abstract static class PartialForwardingClientCall<ReqT, RespT>
1313-
extends ClientCall<ReqT, RespT> {
1314-
protected abstract ClientCall<?, ?> delegate();
13151326

13161327
@Override
13171328
public void request(int numMessages) {
@@ -1320,11 +1331,16 @@ public void request(int numMessages) {
13201331

13211332
@Override
13221333
public void cancel(@Nullable String message, @Nullable Throwable cause) {
1334+
aborted = false;
1335+
cancelCalled = true;
13231336
delegate().cancel(message, cause);
13241337
}
13251338

13261339
@Override
13271340
public void halfClose() {
1341+
if (aborted) {
1342+
return;
1343+
}
13281344
delegate().halfClose();
13291345
}
13301346

‎xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
package io.grpc.xds;
1818

1919
import static com.google.common.truth.Truth.assertThat;
20+
import static org.junit.Assert.assertThrows;
2021

2122
import com.google.common.io.BaseEncoding;
2223
import com.google.common.util.concurrent.MoreExecutors;
@@ -11423,6 +11424,114 @@ public void onClose(Status status, Metadata trailers) {
1142311424
channelManager.close();
1142411425
}
1142511426

11427+
@Test
11428+
@SuppressWarnings("unchecked")
11429+
public void
11430+
givenSendMessageIoException_whenSubsequentSendMessageOrHalfClose_thenNoOp()
11431+
throws Exception {
11432+
ExternalProcessor proto = createBaseProto(extProcServerName)
11433+
.setProcessingMode(ProcessingMode.newBuilder()
11434+
.setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND)
11435+
.setRequestBodyMode(ProcessingMode.BodySendMode.GRPC)
11436+
.setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP)
11437+
.setResponseTrailerMode(ProcessingMode.HeaderSendMode.SKIP)
11438+
.build())
11439+
.build();
11440+
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
11441+
provider.parseFilterConfig(Any.pack(proto), filterContext);
11442+
ExternalProcessorFilterConfig filterConfig = configOrError.config;
11443+
11444+
ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
11445+
new ExternalProcessorGrpc.ExternalProcessorImplBase() {
11446+
@Override
11447+
public StreamObserver<ProcessingRequest> process(
11448+
final StreamObserver<ProcessingResponse> responseObserver) {
11449+
((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
11450+
return new StreamObserver<ProcessingRequest>() {
11451+
@Override
11452+
public void onNext(ProcessingRequest request) {
11453+
if (request.hasRequestHeaders()) {
11454+
responseObserver.onNext(ProcessingResponse.newBuilder()
11455+
.setRequestHeaders(HeadersResponse.newBuilder().build())
11456+
.build());
11457+
}
11458+
}
11459+
11460+
@Override
11461+
public void onError(Throwable t) {}
11462+
11463+
@Override
11464+
public void onCompleted() {
11465+
responseObserver.onCompleted();
11466+
}
11467+
};
11468+
}
11469+
};
11470+
grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName)
11471+
.addService(extProcImpl)
11472+
.directExecutor()
11473+
.build().start());
11474+
11475+
CachedChannelManager channelManager = new CachedChannelManager(config ->
11476+
grpcCleanup.register(
11477+
InProcessChannelBuilder.forName(extProcServerName).directExecutor().build()));
11478+
11479+
ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
11480+
filterConfig, channelManager, scheduler, FAKE_CONTEXT);
11481+
11482+
dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService")
11483+
.addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall(
11484+
(request, responseObserver) -> {
11485+
// No-op
11486+
}))
11487+
.build());
11488+
11489+
ManagedChannel dataPlaneChannel =
11490+
grpcCleanup.register(
11491+
InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build());
11492+
11493+
CallOptions callOptions = DEFAULT_CALL_OPTIONS
11494+
.withExecutor(MoreExecutors.directExecutor())
11495+
.withOption(XdsNameResolver.AUTO_HOST_REWRITE_KEY, false);
11496+
ClientCall<InputStream, InputStream> rawProxyCall =
11497+
(ClientCall<InputStream, InputStream>) (ClientCall<?, ?>)
11498+
interceptor.interceptCall(METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
11499+
11500+
final CountDownLatch closeLatch = new CountDownLatch(1);
11501+
final AtomicReference<Status> closeStatus = new AtomicReference<>();
11502+
rawProxyCall.start(new ClientCall.Listener<InputStream>() {
11503+
@Override
11504+
public void onClose(Status status, Metadata trailers) {
11505+
closeStatus.set(status);
11506+
closeLatch.countDown();
11507+
}
11508+
}, new Metadata());
11509+
11510+
IOException ioException = new IOException("Simulated stream read error");
11511+
InputStream failingStream = new InputStream() {
11512+
@Override
11513+
public int read() throws IOException {
11514+
throw ioException;
11515+
}
11516+
};
11517+
11518+
// Internal cancellation during sendMessage when reading InputStream fails
11519+
rawProxyCall.sendMessage(failingStream);
11520+
11521+
assertThat(closeLatch.await(5, TimeUnit.SECONDS)).isTrue();
11522+
assertThat(closeStatus.get().getCode()).isEqualTo(Status.Code.CANCELLED);
11523+
assertThat(closeStatus.get().getDescription())
11524+
.isEqualTo("Failed to serialize message for External Processor");
11525+
assertThat(closeStatus.get().getCause()).isSameInstanceAs(ioException);
11526+
11527+
// Subsequent sendMessage and halfClose after internal cancellation must be no-ops
11528+
// and not throw IllegalStateException.
11529+
rawProxyCall.sendMessage(new ByteArrayInputStream("ignored".getBytes(StandardCharsets.UTF_8)));
11530+
rawProxyCall.halfClose();
11531+
11532+
channelManager.close();
11533+
}
11534+
1142611535
// --- Category 18: Immediate Response Handling ---
1142711536

1142811537
@Test
@@ -12136,6 +12245,11 @@ public void onCompleted() {
1213612245
// Application cancels the RPC
1213712246
proxyCall.cancel("User cancelled", null);
1213812247

12248+
// Subsequent sendMessage or halfClose after explicit application cancellation must throw
12249+
// IllegalStateException.
12250+
assertThrows(IllegalStateException.class, () -> proxyCall.sendMessage("after-cancel"));
12251+
assertThrows(IllegalStateException.class, () -> proxyCall.halfClose());
12252+
1213912253
// Verify sidecar stream also cancelled
1214012254
assertThat(cancelLatch.await(5, TimeUnit.SECONDS)).isTrue();
1214112255

‎xds/src/test/java/io/grpc/xds/XdsNameResolverTest.java‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import static io.grpc.xds.FaultFilter.HEADER_ABORT_PERCENTAGE_KEY;
2424
import static io.grpc.xds.FaultFilter.HEADER_DELAY_KEY;
2525
import static io.grpc.xds.FaultFilter.HEADER_DELAY_PERCENTAGE_KEY;
26+
import static org.junit.Assert.assertThrows;
2627
import static org.mockito.AdditionalAnswers.delegatesTo;
2728
import static org.mockito.ArgumentMatchers.any;
2829
import static org.mockito.ArgumentMatchers.anyInt;
@@ -75,6 +76,7 @@
7576
import io.grpc.Status;
7677
import io.grpc.Status.Code;
7778
import io.grpc.StatusOr;
79+
import io.grpc.StatusRuntimeException;
7880
import io.grpc.SynchronizationContext;
7981
import io.grpc.inprocess.InProcessChannelBuilder;
8082
import io.grpc.inprocess.InProcessServerBuilder;
@@ -3227,6 +3229,10 @@ public String parse(InputStream stream) {
32273229
ClientCall.Listener<String> listener = mock(ClientCall.Listener.class);
32283230
call.start(listener, new Metadata());
32293231
call.sendMessage("World");
3232+
// Subsequent sendMessage and halfClose after internal sendMessage cancellation are no-ops
3233+
// and must not throw IllegalStateException.
3234+
call.sendMessage("World2");
3235+
call.halfClose();
32303236

32313237
ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
32323238
verify(listener).onClose(statusCaptor.capture(), any(Metadata.class));
@@ -3235,6 +3241,29 @@ public String parse(InputStream stream) {
32353241
assertThat(status.getDescription())
32363242
.isEqualTo("Failed to read message for raw message interceptor");
32373243
assertThat(status.getCause()).isSameInstanceAs(ioException);
3244+
3245+
// Verify ClientCalls.blockingUnaryCall (which calls sendMessage followed by halfClose)
3246+
// surfaces StatusRuntimeException(CANCELLED) rather than IllegalStateException.
3247+
StatusRuntimeException sre = assertThrows(
3248+
StatusRuntimeException.class,
3249+
() -> ClientCalls.blockingUnaryCall(
3250+
interceptedChannel, failingRequestMethod, CallOptions.DEFAULT, "World"));
3251+
assertThat(sre.getStatus().getCode()).isEqualTo(Status.Code.CANCELLED);
3252+
assertThat(sre.getStatus().getDescription())
3253+
.isEqualTo("Failed to read message for raw message interceptor");
3254+
assertThat(sre.getStatus().getCause()).isSameInstanceAs(ioException);
3255+
3256+
// Verify explicit application cancellation still causes subsequent sendMessage and halfClose
3257+
// to throw IllegalStateException.
3258+
ClientCall<String, String> explicitlyCancelledCall =
3259+
interceptedChannel.newCall(failingRequestMethod, CallOptions.DEFAULT);
3260+
@SuppressWarnings("unchecked")
3261+
ClientCall.Listener<String> explicitCancelListener = mock(ClientCall.Listener.class);
3262+
explicitlyCancelledCall.start(explicitCancelListener, new Metadata());
3263+
explicitlyCancelledCall.cancel("Application cancelled", null);
3264+
assertThrows(
3265+
IllegalStateException.class, () -> explicitlyCancelledCall.sendMessage("World"));
3266+
assertThrows(IllegalStateException.class, explicitlyCancelledCall::halfClose);
32383267
}
32393268
}
32403269

0 commit comments

Comments
 (0)