Skip to content

[ISSUE #10890] Validate gRPC client language metadata #10891

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

Open
Aias00 wants to merge 2 commits into
base: develop
Choose a base branch
from
+53 −3
Open

[ISSUE #10890] Validate gRPC client language metadata #10891

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2aa073a
[ISSUE #10890] Validate gRPC client language metadata
Aias00 Aug 10, 2026
2d79a49
[ISSUE #10890] Cover blank client language metadata
Aias00 Aug 10, 2026
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
18 changes: 15 additions & 3 deletions proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ public CompletableFuture<NotifyClientTerminationResponse> notifyClientTerminatio

try {
String clientId = ctx.getClientID();
LanguageCode languageCode = LanguageCode.valueOf(ctx.getLanguage());
LanguageCode languageCode = parseLanguage(ctx);
Settings clientSettings = grpcClientSettingsManager.removeAndGetClientSettings(ctx);
if (clientSettings == null) {
future.complete(NotifyClientTerminationResponse.newBuilder()
Expand Down Expand Up @@ -415,7 +415,7 @@ protected TelemetryCommand processClientSettings(ProxyContext ctx, TelemetryComm

protected GrpcClientChannel registerProducer(ProxyContext ctx, String topicName) {
String clientId = ctx.getClientID();
LanguageCode languageCode = LanguageCode.valueOf(ctx.getLanguage());
LanguageCode languageCode = parseLanguage(ctx);

GrpcClientChannel channel = this.grpcChannelManager.createChannel(ctx, clientId);
// use topic name as producer group
Expand All @@ -431,7 +431,7 @@ protected GrpcClientChannel registerProducer(ProxyContext ctx, String topicName)
protected GrpcClientChannel registerConsumer(ProxyContext ctx, String consumerGroup, ClientType clientType,
List<SubscriptionEntry> subscriptionEntryList, boolean updateSubscription) {
String clientId = ctx.getClientID();
LanguageCode languageCode = LanguageCode.valueOf(ctx.getLanguage());
LanguageCode languageCode = parseLanguage(ctx);

GrpcClientChannel channel = this.grpcChannelManager.createChannel(ctx, clientId);
ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, parseClientVersion(ctx.getClientVersion()));
Expand Down Expand Up @@ -461,6 +461,18 @@ private int parseClientVersion(String clientVersionStr) {
return clientVersion;
}

private LanguageCode parseLanguage(ProxyContext ctx) {
String language = ctx.getLanguage();
if (StringUtils.isBlank(language)) {
throw new GrpcProxyException(Code.BAD_REQUEST, "language cannot be empty");
}
try {
return LanguageCode.valueOf(language);
} catch (IllegalArgumentException e) {
throw new GrpcProxyException(Code.BAD_REQUEST, "unsupported language: " + language, e);
}
}

protected void reportThreadStackTrace(ProxyContext ctx, Status status, ThreadStackTrace request) {
String nonce = request.getNonce();
String threadStack = request.getThreadStackTrace();
Expand Down
38 changes: 38 additions & 0 deletions proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,6 +325,44 @@ public void testErrorProducerConfig() throws Throwable {
}
}

@Test
public void testInvalidLanguageIsReportedAsInvalidArgument() throws Throwable {
ProxyContext context = createContext().setLanguage("UNKNOWN_LANGUAGE");
try {
this.sendClientTelemetry(
context,
Settings.newBuilder()
.setClientType(ClientType.PRODUCER)
.setPublishing(Publishing.newBuilder()
.addTopics(Resource.newBuilder().setName(TOPIC).build())
.build())
.build()).get();
fail();
} catch (ExecutionException e) {
StatusRuntimeException exception = (StatusRuntimeException) e.getCause();
assertEquals(Status.Code.INVALID_ARGUMENT, exception.getStatus().getCode());
}
}

@Test
public void testBlankLanguageIsReportedAsInvalidArgument() throws Throwable {
ProxyContext context = createContext().setLanguage("");
try {
this.sendClientTelemetry(
context,
Settings.newBuilder()
.setClientType(ClientType.PRODUCER)
.setPublishing(Publishing.newBuilder()
.addTopics(Resource.newBuilder().setName(TOPIC).build())
.build())
.build()).get();
fail();
} catch (ExecutionException e) {
StatusRuntimeException exception = (StatusRuntimeException) e.getCause();
assertEquals(Status.Code.INVALID_ARGUMENT, exception.getStatus().getCode());
}
}

@Test
public void testEmptySettings() throws Throwable {
ProxyContext context = createContext();
Expand Down
Loading