Skip to content

Commit 0eedead

Browse files
sushantmanesushantmane
authored
[server][grpc][client] Convert VeniceReadResponseStatus to enum and add GrpcUtils helpers (linkedin#2462)
Convert VeniceReadResponseStatus from a static int constants holder to a proper enum with an int code field and reverse lookup via fromCode(). Add UNKNOWN(-1) sentinel returned when no matching code is found. Add proto serialization helpers in GrpcUtils: - toByteStringNoCopy(byte[]) - toByteString(ByteBuf) Update HttpShortcutResponse to optionally carry a @nullable VeniceReadResponseStatus. Update all server gRPC handlers, client transport layers, and tests to use .getCode() where proto int values are expected. This is PR 1 in a series to rewrite the Venice server gRPC read service, laying the common library foundations for follow-up PRs.
1 parent ae32e8f commit 0eedead

13 files changed

Lines changed: 211 additions & 46 deletions

File tree

‎clients/da-vinci-client/src/test/java/com/linkedin/davinci/blobtransfer/TestP2PFileTransferServerHandler.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,7 @@ public void testRejectTooManyRequest() throws IOException {
147147

148148
if (outbound instanceof FullHttpResponse) {
149149
FullHttpResponse httpResponse = (FullHttpResponse) outbound;
150-
if (httpResponse.status().code() == TOO_MANY_REQUESTS) {
150+
if (httpResponse.status().code() == TOO_MANY_REQUESTS.getCode()) {
151151
foundTooManyRequestsResponse = true;
152152
break;
153153
}

‎clients/venice-client/src/main/java/com/linkedin/venice/fastclient/transport/GrpcTransportClient.java‎

Lines changed: 11 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -246,7 +246,7 @@ public VeniceGrpcStreamObserver(CompletableFuture<TransportClientResponse> respo
246246

247247
@Override
248248
public void onNext(VeniceServerResponse value) {
249-
if (value.getErrorCode() != VeniceReadResponseStatus.OK) {
249+
if (value.getErrorCode() != VeniceReadResponseStatus.OK.getCode()) {
250250
handleResponseError(value);
251251
return;
252252
}
@@ -288,21 +288,16 @@ void handleResponseError(VeniceServerResponse response) {
288288
String errorMessage = response.getErrorMessage();
289289
Exception exception;
290290

291-
switch (statusCode) {
292-
case VeniceReadResponseStatus.BAD_REQUEST:
293-
exception = new VeniceClientHttpException(errorMessage, statusCode);
294-
break;
295-
case VeniceReadResponseStatus.TOO_MANY_REQUESTS:
296-
exception = new VeniceClientRateExceededException(errorMessage);
297-
break;
298-
case VeniceReadResponseStatus.KEY_NOT_FOUND:
299-
exception = null;
300-
break;
301-
default:
302-
exception = new VeniceClientException(
303-
String
304-
.format("An unexpected error occurred with status code: %d, message: %s", statusCode, errorMessage));
305-
break;
291+
VeniceReadResponseStatus responseStatus = VeniceReadResponseStatus.fromCode(statusCode);
292+
if (responseStatus == VeniceReadResponseStatus.BAD_REQUEST) {
293+
exception = new VeniceClientHttpException(errorMessage, statusCode);
294+
} else if (responseStatus == VeniceReadResponseStatus.TOO_MANY_REQUESTS) {
295+
exception = new VeniceClientRateExceededException(errorMessage);
296+
} else if (responseStatus == VeniceReadResponseStatus.KEY_NOT_FOUND) {
297+
exception = null;
298+
} else {
299+
exception = new VeniceClientException(
300+
String.format("An unexpected error occurred with status code: %d, message: %s", statusCode, errorMessage));
306301
}
307302

308303
if (exception != null) {

‎internal/venice-common/src/main/java/com/linkedin/venice/grpc/GrpcUtils.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package com.linkedin.venice.grpc;
22

3+
import com.google.protobuf.ByteString;
4+
import com.google.protobuf.UnsafeByteOperations;
35
import com.linkedin.venice.acl.handler.AccessResult;
46
import com.linkedin.venice.client.exceptions.VeniceClientException;
57
import com.linkedin.venice.exceptions.VeniceException;
@@ -12,6 +14,7 @@
1214
import io.grpc.ServerCall;
1315
import io.grpc.Status;
1416
import io.grpc.TlsChannelCredentials;
17+
import io.netty.buffer.ByteBuf;
1518
import java.io.IOException;
1619
import java.io.InputStream;
1720
import java.nio.file.Files;
@@ -102,6 +105,31 @@ private static KeyStore loadStore(String path, char[] password, String type)
102105
return keyStore;
103106
}
104107

108+
/**
109+
* Wraps a byte array into a {@link ByteString} <b>without copying</b>. The returned {@code ByteString} directly
110+
* aliases the provided array, so the caller <b>must not</b> modify the array after calling this method.
111+
* Violating this contract can cause silent data corruption in serialized gRPC payloads.
112+
*
113+
* <p>If the caller cannot guarantee immutability of the input array, use {@link ByteString#copyFrom(byte[])}
114+
* instead.
115+
*/
116+
public static ByteString toByteStringNoCopy(byte[] bytes) {
117+
if (bytes == null || bytes.length == 0) {
118+
return ByteString.EMPTY;
119+
}
120+
return UnsafeByteOperations.unsafeWrap(bytes);
121+
}
122+
123+
/** Copies readable bytes from a {@link ByteBuf} into a {@link ByteString}. */
124+
public static ByteString toByteString(ByteBuf buf) {
125+
if (buf == null || buf.readableBytes() == 0) {
126+
return ByteString.EMPTY;
127+
}
128+
byte[] bytes = new byte[buf.readableBytes()];
129+
buf.getBytes(buf.readerIndex(), bytes);
130+
return ByteString.copyFrom(bytes);
131+
}
132+
105133
public static ChannelCredentials buildChannelCredentials(SSLFactory sslFactory) {
106134
// TODO: Evaluate if this needs to fail instead since it depends on plain text support on server
107135
if (sslFactory == null) {

‎internal/venice-common/src/main/java/com/linkedin/venice/listener/response/HttpShortcutResponse.java‎

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,24 +1,31 @@
11
package com.linkedin.venice.listener.response;
22

3+
import com.linkedin.venice.response.VeniceReadResponseStatus;
34
import io.netty.handler.codec.http.HttpResponseStatus;
5+
import javax.annotation.Nullable;
46

57

6-
/**
7-
* Created by mwise on 3/11/16.
8-
*/
8+
/** A response object carrying an HTTP status and optional Venice read response status. */
99
public class HttpShortcutResponse {
1010
private final String message;
1111
private final HttpResponseStatus status;
12+
@Nullable
13+
private final VeniceReadResponseStatus veniceReadResponseStatus;
1214

1315
private boolean misroutedStoreVersion = false;
1416

1517
public HttpShortcutResponse(String message, HttpResponseStatus status) {
16-
this.message = message;
17-
this.status = status;
18+
this(message, status, null);
1819
}
1920

2021
public HttpShortcutResponse(HttpResponseStatus status) {
21-
this("", status);
22+
this("", status, null);
23+
}
24+
25+
public HttpShortcutResponse(String message, HttpResponseStatus status, VeniceReadResponseStatus readResponseStatus) {
26+
this.message = message;
27+
this.status = status;
28+
this.veniceReadResponseStatus = readResponseStatus;
2229
}
2330

2431
public String getMessage() {
@@ -29,6 +36,11 @@ public HttpResponseStatus getStatus() {
2936
return status;
3037
}
3138

39+
@Nullable
40+
public VeniceReadResponseStatus getVeniceReadResponseStatus() {
41+
return veniceReadResponseStatus;
42+
}
43+
3244
public boolean isMisroutedStoreVersion() {
3345
return misroutedStoreVersion;
3446
}

‎internal/venice-common/src/main/java/com/linkedin/venice/response/VeniceReadResponseStatus.java‎

Lines changed: 41 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,49 @@
11
package com.linkedin.venice.response;
22

3+
import java.util.Collections;
4+
import java.util.HashMap;
5+
import java.util.Map;
6+
7+
38
/**
49
* Enumeration of response status codes for Venice read requests.
510
* <p>
6-
* **Positive values** correspond to standard HTTP status codes and can be used directly in HTTP responses.
7-
* **Negative values** represent custom Venice-specific error codes.
11+
* Positive values correspond to standard HTTP status codes and can be used directly in HTTP responses.
12+
* Negative values represent custom Venice-specific error codes.
813
* <p>
9-
* For example, a status code of `200` indicates a successful read, while a status code of `-100` might indicate a specific Venice-related error.
14+
* {@link #UNKNOWN} is a sentinel value (code {@code -1}) returned by {@link #fromCode(int)} when no matching status
15+
* is found. The code {@code -1} is reserved and must never be used as a real status code.
1016
*/
11-
public class VeniceReadResponseStatus {
12-
public static final int KEY_NOT_FOUND = -420;
13-
14-
public static final int OK = 200;
15-
public static final int BAD_REQUEST = 400;
16-
public static final int INTERNAL_ERROR = 500;
17-
public static final int TOO_MANY_REQUESTS = 429;
18-
public static final int SERVICE_UNAVAILABLE = 503;
17+
public enum VeniceReadResponseStatus {
18+
UNKNOWN(-1), KEY_NOT_FOUND(-420), OK(200), BAD_REQUEST(400), TOO_MANY_REQUESTS(429), INTERNAL_ERROR(500),
19+
SERVICE_UNAVAILABLE(503);
20+
21+
private final int code;
22+
23+
private static final Map<Integer, VeniceReadResponseStatus> CODE_MAP;
24+
25+
static {
26+
Map<Integer, VeniceReadResponseStatus> map = new HashMap<>();
27+
for (VeniceReadResponseStatus status: values()) {
28+
if (status != UNKNOWN) {
29+
map.put(status.code, status);
30+
}
31+
}
32+
CODE_MAP = Collections.unmodifiableMap(map);
33+
}
34+
35+
VeniceReadResponseStatus(int code) {
36+
this.code = code;
37+
}
38+
39+
public int getCode() {
40+
return code;
41+
}
42+
43+
/**
44+
* Returns the {@link VeniceReadResponseStatus} for the given integer code, or {@link #UNKNOWN} if no match is found.
45+
*/
46+
public static VeniceReadResponseStatus fromCode(int code) {
47+
return CODE_MAP.getOrDefault(code, UNKNOWN);
48+
}
1949
}

‎internal/venice-common/src/test/java/com/linkedin/venice/grpc/GrpcUtilsTest.java‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,11 @@
44
import static org.mockito.Mockito.when;
55
import static org.testng.Assert.assertEquals;
66
import static org.testng.Assert.assertNotNull;
7+
import static org.testng.Assert.assertSame;
78
import static org.testng.Assert.assertTrue;
89
import static org.testng.Assert.expectThrows;
910

11+
import com.google.protobuf.ByteString;
1012
import com.linkedin.venice.acl.handler.AccessResult;
1113
import com.linkedin.venice.client.exceptions.VeniceClientException;
1214
import com.linkedin.venice.exceptions.VeniceException;
@@ -19,6 +21,8 @@
1921
import io.grpc.ServerCall;
2022
import io.grpc.Status;
2123
import io.grpc.TlsChannelCredentials;
24+
import io.netty.buffer.ByteBuf;
25+
import io.netty.buffer.Unpooled;
2226
import java.security.cert.Certificate;
2327
import java.security.cert.X509Certificate;
2428
import javax.net.ssl.KeyManager;
@@ -205,4 +209,43 @@ public void testExtractGrpcClientCertWithEmptyPeerCertificates() throws SSLPeerU
205209
// Verify the exception is thrown
206210
assertNotNull(thrownException);
207211
}
212+
213+
@Test
214+
public void testToByteStringNoCopyFromByteArray() {
215+
byte[] data = new byte[] { 1, 2, 3, 4, 5 };
216+
ByteString result = GrpcUtils.toByteStringNoCopy(data);
217+
assertEquals(result.size(), 5);
218+
assertEquals(result.toByteArray(), data);
219+
}
220+
221+
@Test
222+
public void testToByteStringNoCopyFromNullByteArray() {
223+
assertSame(GrpcUtils.toByteStringNoCopy(null), ByteString.EMPTY);
224+
}
225+
226+
@Test
227+
public void testToByteStringNoCopyFromEmptyByteArray() {
228+
assertSame(GrpcUtils.toByteStringNoCopy(new byte[0]), ByteString.EMPTY);
229+
}
230+
231+
@Test
232+
public void testToByteStringFromByteBuf() {
233+
byte[] data = new byte[] { 10, 20, 30 };
234+
ByteBuf buf = Unpooled.wrappedBuffer(data);
235+
ByteString result = GrpcUtils.toByteString(buf);
236+
assertEquals(result.size(), 3);
237+
assertEquals(result.toByteArray(), data);
238+
buf.release();
239+
}
240+
241+
@Test
242+
public void testToByteStringFromNullByteBuf() {
243+
assertSame(GrpcUtils.toByteString((ByteBuf) null), ByteString.EMPTY);
244+
}
245+
246+
@Test
247+
public void testToByteStringFromEmptyByteBuf() {
248+
ByteBuf buf = Unpooled.EMPTY_BUFFER;
249+
assertSame(GrpcUtils.toByteString(buf), ByteString.EMPTY);
250+
}
208251
}

‎internal/venice-common/src/test/java/com/linkedin/venice/response/VeniceReadResponseStatusTest.java‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
package com.linkedin.venice.response;
2+
3+
import static org.testng.Assert.assertEquals;
4+
import static org.testng.Assert.assertNotNull;
5+
6+
import java.util.HashSet;
7+
import java.util.Set;
8+
import org.testng.annotations.Test;
9+
10+
11+
public class VeniceReadResponseStatusTest {
12+
@Test
13+
public void testGetCode() {
14+
assertEquals(VeniceReadResponseStatus.UNKNOWN.getCode(), -1);
15+
assertEquals(VeniceReadResponseStatus.KEY_NOT_FOUND.getCode(), -420);
16+
assertEquals(VeniceReadResponseStatus.OK.getCode(), 200);
17+
assertEquals(VeniceReadResponseStatus.BAD_REQUEST.getCode(), 400);
18+
assertEquals(VeniceReadResponseStatus.TOO_MANY_REQUESTS.getCode(), 429);
19+
assertEquals(VeniceReadResponseStatus.INTERNAL_ERROR.getCode(), 500);
20+
assertEquals(VeniceReadResponseStatus.SERVICE_UNAVAILABLE.getCode(), 503);
21+
}
22+
23+
@Test
24+
public void testFromCode() {
25+
for (VeniceReadResponseStatus status: VeniceReadResponseStatus.values()) {
26+
if (status == VeniceReadResponseStatus.UNKNOWN) {
27+
continue;
28+
}
29+
VeniceReadResponseStatus resolved = VeniceReadResponseStatus.fromCode(status.getCode());
30+
assertNotNull(resolved, "fromCode should resolve " + status.name());
31+
assertEquals(resolved, status);
32+
}
33+
}
34+
35+
@Test
36+
public void testFromCodeWithUnknownCode() {
37+
assertEquals(VeniceReadResponseStatus.fromCode(999), VeniceReadResponseStatus.UNKNOWN);
38+
assertEquals(VeniceReadResponseStatus.fromCode(0), VeniceReadResponseStatus.UNKNOWN);
39+
assertEquals(VeniceReadResponseStatus.fromCode(-1), VeniceReadResponseStatus.UNKNOWN);
40+
}
41+
42+
@Test
43+
public void testFromCodeWithNegativeCode() {
44+
VeniceReadResponseStatus status = VeniceReadResponseStatus.fromCode(-420);
45+
assertNotNull(status);
46+
assertEquals(status, VeniceReadResponseStatus.KEY_NOT_FOUND);
47+
}
48+
49+
@Test
50+
public void testAllCodesAreUnique() {
51+
Set<Integer> codes = new HashSet<>();
52+
for (VeniceReadResponseStatus status: VeniceReadResponseStatus.values()) {
53+
boolean added = codes.add(status.getCode());
54+
assertEquals(added, true, "Duplicate code found: " + status.getCode());
55+
}
56+
}
57+
}

‎services/venice-server/src/main/java/com/linkedin/venice/listener/grpc/VeniceReadServiceImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ public void batchGet(VeniceClientRequest request, StreamObserver<VeniceServerRes
3131

3232
private void handleRequest(VeniceClientRequest request, StreamObserver<VeniceServerResponse> responseObserver) {
3333
VeniceServerResponse.Builder responseBuilder =
34-
VeniceServerResponse.newBuilder().setErrorCode(VeniceReadResponseStatus.OK);
34+
VeniceServerResponse.newBuilder().setErrorCode(VeniceReadResponseStatus.OK.getCode());
3535
GrpcRequestContext ctx = new GrpcRequestContext(request, responseBuilder, responseObserver);
3636
requestProcessor.process(ctx);
3737
}

‎services/venice-server/src/main/java/com/linkedin/venice/listener/grpc/handlers/GrpcOutboundResponseHandler.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ public void processRequest(GrpcRequestContext ctx) {
5353
ctx.setError();
5454
statsContext.setResponseStatus(NOT_FOUND);
5555
veniceServerResponseBuilder.setData(ByteString.EMPTY);
56-
veniceServerResponseBuilder.setErrorCode(VeniceReadResponseStatus.KEY_NOT_FOUND);
56+
veniceServerResponseBuilder.setErrorCode(VeniceReadResponseStatus.KEY_NOT_FOUND.getCode());
5757
veniceServerResponseBuilder.setErrorMessage("Key not found");
5858
invokeNextHandler(ctx);
5959
}

‎services/venice-server/src/main/java/com/linkedin/venice/listener/grpc/handlers/GrpcReadQuotaEnforcementHandler.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,15 +30,15 @@ public void processRequest(GrpcRequestContext context) {
3030
context.setError();
3131
if (result == ReadQuotaEnforcementHandler.QuotaEnforcementResult.BAD_REQUEST) {
3232
context.getVeniceServerResponseBuilder()
33-
.setErrorCode(VeniceReadResponseStatus.BAD_REQUEST)
33+
.setErrorCode(VeniceReadResponseStatus.BAD_REQUEST.getCode())
3434
.setErrorMessage(INVALID_REQUEST_RESOURCE_MSG + request.getResourceName());
3535
} else if (result == ReadQuotaEnforcementHandler.QuotaEnforcementResult.REJECTED) {
3636
context.getVeniceServerResponseBuilder()
37-
.setErrorCode(VeniceReadResponseStatus.TOO_MANY_REQUESTS)
37+
.setErrorCode(VeniceReadResponseStatus.TOO_MANY_REQUESTS.getCode())
3838
.setErrorMessage("");
3939
} else if (result == ReadQuotaEnforcementHandler.QuotaEnforcementResult.OVER_CAPACITY) {
4040
context.getVeniceServerResponseBuilder()
41-
.setErrorCode(VeniceReadResponseStatus.SERVICE_UNAVAILABLE)
41+
.setErrorCode(VeniceReadResponseStatus.SERVICE_UNAVAILABLE.getCode())
4242
.setErrorMessage(SERVER_OVER_CAPACITY_MSG);
4343
}
4444

0 commit comments

Comments
 (0)