Skip to content

feat: Skip gRPC trailers when last field is set. #268

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
ath-08 merged 3 commits into from
Dec 16, 2025
Merged

feat: Skip gRPC trailers when last field is set. #268

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6546208
feat: Skip gRPC trailers when `last` field is set.
ath-08 Dec 13, 2025
6679da8
upgrade versions to match latest spanneradapter
ath-08 Dec 13, 2025
746d1d2
Address code review comments
ath-08 Dec 15, 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
4 changes: 4 additions & 0 deletions ...panner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterClientWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,10 @@ ByteString sendGrpcRequest(
for (AdaptMessageResponse adaptMessageResponse : serverStream) {
adaptMessageResponse.getStateUpdatesMap().forEach(attachmentsCache::put);
collectedPayloads.add(adaptMessageResponse.getPayload());
if (adaptMessageResponse.getLast()) {
// This is the last message in the stream. Stop processing immediately.
break;
}
}
} catch (RuntimeException e) {
LOG.error("Error executing AdaptMessage request: ", e);
Expand Down
73 changes: 55 additions & 18 deletions ...er-cassandra/src/test/java/com/google/cloud/spanner/adapter/AdapterClientWrapperTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,11 +74,7 @@ public void sendGrpcRequest_SuccessfulResponse() {
Map<String, String> stateUpdates = new HashMap<>();
stateUpdates.put("k1", "v1");
stateUpdates.put("k2", "v2");
AdaptMessageResponse mockResponse =
AdaptMessageResponse.newBuilder()
.setPayload(ByteString.copyFromUtf8("test response"))
.putAllStateUpdates(stateUpdates)
.build();
AdaptMessageResponse mockResponse = createAdaptMessageResponse("test response", stateUpdates);
Iterator<AdaptMessageResponse> mockResponseIterator =
Collections.singletonList(mockResponse).iterator();
AdaptMessageRequest expectedRequest =
Expand Down Expand Up @@ -106,21 +102,13 @@ public void sendGrpcRequest_MultipleResponses() {
stateUpdates1.put("k1", "v1");
stateUpdates1.put("k2", "v2");
AdaptMessageResponse mockResponse1 =
AdaptMessageResponse.newBuilder()
.setPayload(ByteString.copyFromUtf8(" test response 1"))
.putAllStateUpdates(stateUpdates1)
.build();
createAdaptMessageResponse(" test response 1", stateUpdates1);
Map<String, String> stateUpdates2 = new HashMap<>();
stateUpdates2.put("k3", "v3");
AdaptMessageResponse mockResponse2 =
AdaptMessageResponse.newBuilder()
.setPayload(ByteString.copyFromUtf8(" test response 2"))
.putAllStateUpdates(stateUpdates2)
.build();
createAdaptMessageResponse(" test response 2", stateUpdates2);
AdaptMessageResponse mockResponse3 =
AdaptMessageResponse.newBuilder()
.setPayload(ByteString.copyFromUtf8("test header"))
.build();
createAdaptMessageResponse("test header", Collections.emptyMap());
Iterator<AdaptMessageResponse> mockResponseIterator =
Arrays.asList(mockResponse1, mockResponse2, mockResponse3).iterator();
when(mockServerStream.iterator()).thenReturn(mockResponseIterator);
Expand All @@ -142,6 +130,42 @@ public void sendGrpcRequest_MultipleResponses() {
assertThat(attachmentsCache.get("k3")).hasValue("v3");
}

@Test
public void sendGrpcRequest_MultipleResponsesWithLastBitSet() {
int streamId = 1;
byte[] payload = "test payload".getBytes();
Map<String, String> stateUpdates1 = new HashMap<>();
stateUpdates1.put("k1", "v1");
stateUpdates1.put("k2", "v2");
AdaptMessageResponse mockResponse1 =
createAdaptMessageResponse(" test response 1", stateUpdates1);
Map<String, String> stateUpdates2 = new HashMap<>();
stateUpdates2.put("k3", "v3");
AdaptMessageResponse mockResponse2 =
Comment thread
ath-08 marked this conversation as resolved.
createAdaptMessageResponse(" test response 2", stateUpdates2);
AdaptMessageResponse mockResponse3 =
createAdaptMessageResponse("test header (last)", Collections.emptyMap(), true);
Iterator<AdaptMessageResponse> mockResponseIterator =
Arrays.asList(mockResponse1, mockResponse2, mockResponse3).iterator();
when(mockServerStream.iterator()).thenReturn(mockResponseIterator);
AdaptMessageRequest expectedRequest =
AdaptMessageRequest.newBuilder()
.setName("test-session")
.setProtocol("cassandra")
.setPayload(ByteString.copyFrom(payload))
.build();

ByteString response =
adapterClientWrapper.sendGrpcRequest(payload, new HashMap<>(), context, streamId);

verify(mockCallable).call(expectedRequest, context);
assertThat(response)
.isEqualTo(ByteString.copyFromUtf8("test header (last) test response 1 test response 2"));
assertThat(attachmentsCache.get("k1")).hasValue("v1");
Comment thread
ath-08 marked this conversation as resolved.
assertThat(attachmentsCache.get("k2")).hasValue("v2");
assertThat(attachmentsCache.get("k3")).hasValue("v3");
}

@Test
public void sendGrpcRequest_NoResponse() {
int streamId = 1;
Expand All @@ -156,8 +180,7 @@ public void sendGrpcRequest_NoResponse() {
when(mockServerStream.iterator()).thenReturn(mockResponseIterator);
when(mockSession.getName()).thenReturn("test-session");

ByteString response =
adapterClientWrapper.sendGrpcRequest(payload, new HashMap<>(), context, streamId);
adapterClientWrapper.sendGrpcRequest(payload, new HashMap<>(), context, streamId);

verify(mockCallable).call(expectedRequest, context);
}
Expand All @@ -174,4 +197,18 @@ public void sendGrpcRequest_SessionCreationFailure() {
adapterClientWrapper.sendGrpcRequest(
payload, new HashMap<>(), GrpcCallContext.createDefault(), streamId));
}

private static AdaptMessageResponse createAdaptMessageResponse(
String text, Map<String, String> stateUpdates, boolean isLast) {
return AdaptMessageResponse.newBuilder()
.setPayload(ByteString.copyFromUtf8(text))
.putAllStateUpdates(stateUpdates)
.setLast(isLast)
.build();
}

private static AdaptMessageResponse createAdaptMessageResponse(
String text, Map<String, String> stateUpdates) {
return createAdaptMessageResponse(text, stateUpdates, false);
}
}
8 changes: 4 additions & 4 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
<parent>
<groupId>com.google.cloud</groupId>
<artifactId>sdk-platform-java-config</artifactId>
<version>3.47.0</version>
<version>3.54.2</version>
</parent>
<modelVersion>4.0.0</modelVersion>

Expand Down Expand Up @@ -79,7 +79,7 @@
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>google-cloud-spanneradapter</artifactId>
<version>0.0.1</version>
<version>0.16.0</version>
</dependency>
<dependency>
<groupId>com.google.cloud</groupId>
Expand Down Expand Up @@ -110,12 +110,12 @@
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-api</artifactId>
<version>1.47.0</version>
<version>1.52.0</version>
Comment thread
ath-08 marked this conversation as resolved.
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-sdk</artifactId>
<version>1.47.0</version>
<version>1.52.0</version>
</dependency>
<dependency>
<groupId>com.google.cloud.opentelemetry</groupId>
Expand Down