Copilot commented on code in PR #10796:
URL: https://github.com/apache/rocketmq/pull/10796#discussion_r3704895502
##########
proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java:
##########
@@ -158,8 +163,36 @@ protected RemotingCommand
unregisterClient(ChannelHandlerContext ctx, RemotingCo
protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx,
RemotingCommand request,
ProxyContext context) {
final RemotingCommand response =
RemotingCommand.createResponseCommand(null);
+ CheckClientRequestBody requestBody =
CheckClientRequestBody.decode(request.getBody(),
+ CheckClientRequestBody.class);
+ if (requestBody != null && requestBody.getSubscriptionData() != null) {
+ SubscriptionData subscriptionData =
requestBody.getSubscriptionData();
+ if
(ExpressionType.isTagType(subscriptionData.getExpressionType())) {
+ response.setCode(ResponseCode.SUCCESS);
+ response.setRemark(null);
+ return response;
+ }
+
+ if
(!ConfigurationManager.getProxyConfig().isEnablePropertyFilter()) {
+ response.setCode(ResponseCode.SYSTEM_ERROR);
+ response.setRemark("The proxy does not support consumer to
filter message by "
+ + subscriptionData.getExpressionType());
+ return response;
+ }
Review Comment:
The remark message is grammatically incorrect and doesn’t provide an
actionable next step. Consider rewriting it to clearly state that SQL/property
filtering is disabled and how to enable it (e.g., reference
`enablePropertyFilter`) while still including the expression type.
##########
proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java:
##########
@@ -158,8 +163,36 @@ protected RemotingCommand
unregisterClient(ChannelHandlerContext ctx, RemotingCo
protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx,
RemotingCommand request,
ProxyContext context) {
final RemotingCommand response =
RemotingCommand.createResponseCommand(null);
+ CheckClientRequestBody requestBody =
CheckClientRequestBody.decode(request.getBody(),
+ CheckClientRequestBody.class);
+ if (requestBody != null && requestBody.getSubscriptionData() != null) {
+ SubscriptionData subscriptionData =
requestBody.getSubscriptionData();
+ if
(ExpressionType.isTagType(subscriptionData.getExpressionType())) {
+ response.setCode(ResponseCode.SUCCESS);
+ response.setRemark(null);
+ return response;
+ }
+
+ if
(!ConfigurationManager.getProxyConfig().isEnablePropertyFilter()) {
+ response.setCode(ResponseCode.SYSTEM_ERROR);
+ response.setRemark("The proxy does not support consumer to
filter message by "
+ + subscriptionData.getExpressionType());
+ return response;
+ }
+
+ try {
+
FilterFactory.INSTANCE.get(subscriptionData.getExpressionType()).compile(subscriptionData.getSubString());
+ } catch (Exception e) {
+ log.warn("Client {}@{} filter message, but failed to compile
expression! sub={}, error={}",
+ requestBody.getClientId(), requestBody.getGroup(),
requestBody.getSubscriptionData(), e.getMessage());
+ response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
+ response.setRemark(e.getMessage());
+ return response;
Review Comment:
The warning log drops the exception stack trace (only logs
`e.getMessage()`), which makes production triage harder—especially since
`getMessage()` can be null (e.g., some `NullPointerException`s). Log the
exception itself (pass `e` to the logger) so you retain stack trace and root
cause details.
##########
proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.proxy.remoting.activity;
+
+import org.apache.rocketmq.common.filter.ExpressionType;
+import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.config.InitConfigTest;
+import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.proxy.remoting.channel.RemotingChannelManager;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.CheckClientRequestBody;
+import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ClientManagerActivityTest extends InitConfigTest {
+
+ private ClientManagerActivity clientManagerActivity;
+ @Mock
+ private MessagingProcessor messagingProcessor;
+ @Mock
+ private RemotingChannelManager remotingChannelManager;
+
+ @Before
+ public void setUp() {
+ this.clientManagerActivity = new ClientManagerActivity(null,
messagingProcessor, remotingChannelManager);
+ }
+
+ @Test
+ public void testCheckClientConfigWithTagExpression() {
+ RemotingCommand response =
clientManagerActivity.checkClientConfig(null,
+ createRequest(ExpressionType.TAG, "tagA || tagB"),
ProxyContext.create());
+
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ }
+
+ @Test
+ public void testCheckClientConfigRejectsPropertyFilterWhenDisabled() {
+ RemotingCommand response =
clientManagerActivity.checkClientConfig(null,
+ createRequest(ExpressionType.SQL92, "a is not null"),
ProxyContext.create());
+
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
+ assertThat(response.getRemark()).contains(ExpressionType.SQL92);
+ }
+
+ @Test
+ public void testCheckClientConfigRejectsInvalidPropertyFilterExpression() {
+ ConfigurationManager.getProxyConfig().setEnablePropertyFilter(true);
+
+ RemotingCommand response =
clientManagerActivity.checkClientConfig(null,
+ createRequest(ExpressionType.SQL92, "a = "),
ProxyContext.create());
+
+
assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
+ }
Review Comment:
This test mutates global config (`enablePropertyFilter`) but never resets
it, which can make the suite order-dependent/flaky (e.g., if another test
assumes the default `false` and runs after this one). Save the old value and
restore it in a finally block, or reset the config in `@After`/`@Before`.
##########
proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java:
##########
@@ -158,8 +163,36 @@ protected RemotingCommand
unregisterClient(ChannelHandlerContext ctx, RemotingCo
protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx,
RemotingCommand request,
ProxyContext context) {
final RemotingCommand response =
RemotingCommand.createResponseCommand(null);
+ CheckClientRequestBody requestBody =
CheckClientRequestBody.decode(request.getBody(),
+ CheckClientRequestBody.class);
+ if (requestBody != null && requestBody.getSubscriptionData() != null) {
+ SubscriptionData subscriptionData =
requestBody.getSubscriptionData();
+ if
(ExpressionType.isTagType(subscriptionData.getExpressionType())) {
+ response.setCode(ResponseCode.SUCCESS);
+ response.setRemark(null);
+ return response;
+ }
+
+ if
(!ConfigurationManager.getProxyConfig().isEnablePropertyFilter()) {
+ response.setCode(ResponseCode.SYSTEM_ERROR);
+ response.setRemark("The proxy does not support consumer to
filter message by "
+ + subscriptionData.getExpressionType());
+ return response;
+ }
+
+ try {
+
FilterFactory.INSTANCE.get(subscriptionData.getExpressionType()).compile(subscriptionData.getSubString());
+ } catch (Exception e) {
+ log.warn("Client {}@{} filter message, but failed to compile
expression! sub={}, error={}",
+ requestBody.getClientId(), requestBody.getGroup(),
requestBody.getSubscriptionData(), e.getMessage());
+ response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
+ response.setRemark(e.getMessage());
+ return response;
+ }
Review Comment:
The updated behavior includes a success path for valid SQL/property filter
compilation when `enablePropertyFilter` is enabled, but the tests only cover
invalid compilation. Add a regression test that enables `enablePropertyFilter`,
sends a valid SQL92 expression, and asserts `ResponseCode.SUCCESS` (and
expected remark behavior) to prevent regressions in the allowed path.
##########
proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java:
##########
@@ -158,8 +163,36 @@ protected RemotingCommand
unregisterClient(ChannelHandlerContext ctx, RemotingCo
protected RemotingCommand checkClientConfig(ChannelHandlerContext ctx,
RemotingCommand request,
ProxyContext context) {
final RemotingCommand response =
RemotingCommand.createResponseCommand(null);
+ CheckClientRequestBody requestBody =
CheckClientRequestBody.decode(request.getBody(),
+ CheckClientRequestBody.class);
+ if (requestBody != null && requestBody.getSubscriptionData() != null) {
Review Comment:
`CheckClientRequestBody.decode(request.getBody(), ...)` is called without
guarding against a null body or decode failure. If `request.getBody()` is null
or contains invalid payload, this can throw and fail the request handling path.
Wrap decoding in a try/catch (or null-check `request.getBody()` first) and
return an error `RemotingCommand` (with an appropriate `ResponseCode` + remark)
instead of throwing.
##########
proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivityTest.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.proxy.remoting.activity;
+
+import org.apache.rocketmq.common.filter.ExpressionType;
+import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.config.InitConfigTest;
+import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.proxy.remoting.channel.RemotingChannelManager;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.CheckClientRequestBody;
+import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ClientManagerActivityTest extends InitConfigTest {
+
+ private ClientManagerActivity clientManagerActivity;
+ @Mock
+ private MessagingProcessor messagingProcessor;
+ @Mock
+ private RemotingChannelManager remotingChannelManager;
+
+ @Before
+ public void setUp() {
+ this.clientManagerActivity = new ClientManagerActivity(null,
messagingProcessor, remotingChannelManager);
+ }
Review Comment:
This test mutates global config (`enablePropertyFilter`) but never resets
it, which can make the suite order-dependent/flaky (e.g., if another test
assumes the default `false` and runs after this one). Save the old value and
restore it in a finally block, or reset the config in `@After`/`@Before`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
