Copilot commented on code in PR #10798: URL: https://github.com/apache/rocketmq/pull/10798#discussion_r3704931584
########## proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/TransactionActivityTest.java: ########## @@ -0,0 +1,127 @@ +/* + * 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 io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import java.util.concurrent.CompletableFuture; +import org.apache.rocketmq.client.exception.MQBrokerException; +import org.apache.rocketmq.common.MQVersion; +import org.apache.rocketmq.common.sysflag.MessageSysFlag; +import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.config.InitConfigTest; +import org.apache.rocketmq.proxy.processor.MessagingProcessor; +import org.apache.rocketmq.proxy.processor.TransactionStatus; +import org.apache.rocketmq.proxy.service.channel.SimpleChannel; +import org.apache.rocketmq.proxy.service.channel.SimpleChannelHandlerContext; +import org.apache.rocketmq.remoting.common.RemotingHelper; +import org.apache.rocketmq.remoting.netty.AttributeKeys; +import org.apache.rocketmq.remoting.protocol.LanguageCode; +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.header.EndTransactionRequestHeader; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class TransactionActivityTest extends InitConfigTest { + + private TransactionActivity transactionActivity; + @Mock + private MessagingProcessor messagingProcessor; + @Spy + private ChannelHandlerContext ctx = new SimpleChannelHandlerContext(new SimpleChannel(null, "0.0.0.0:0", "1.1.1.1:1")) { + @Override + public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) { + return null; + } + }; Review Comment: `ChannelHandlerContext.writeAndFlush(...)` is expected to return a non-null `ChannelFuture`. Returning `null` can break code under test if it adds listeners or otherwise interacts with the future (leading to NPEs) and can make tests brittle as implementation details change. Return a successfully-completed promise/future (e.g., mark `promise` success and return it) to satisfy the contract. ########## proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/TransactionActivity.java: ########## @@ -63,7 +63,11 @@ protected RemotingCommand processRequest0(ChannelHandlerContext ctx, RemotingCom requestHeader.getProducerGroup(), transactionStatus, requestHeader.getFromTransactionCheck() - ); - return response; + ).thenRun(() -> writeResponse(ctx, context, request, response)) + .exceptionally(t -> { + writeErrResponse(ctx, context, request, t); + return null; + }); Review Comment: `exceptionally` receives the stage failure, which is often a `CompletionException` wrapper. Passing the wrapper through to `writeErrResponse(...)` can cause error mapping to miss the underlying cause (e.g., `MQBrokerException`) and return the wrong response code/remark. Unwrap the throwable before mapping (e.g., use `t.getCause()` when present) so the existing remoting error response mapping sees the real exception. ########## proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/TransactionActivityTest.java: ########## @@ -0,0 +1,127 @@ +/* + * 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 io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import java.util.concurrent.CompletableFuture; +import org.apache.rocketmq.client.exception.MQBrokerException; +import org.apache.rocketmq.common.MQVersion; +import org.apache.rocketmq.common.sysflag.MessageSysFlag; +import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.config.InitConfigTest; +import org.apache.rocketmq.proxy.processor.MessagingProcessor; +import org.apache.rocketmq.proxy.processor.TransactionStatus; +import org.apache.rocketmq.proxy.service.channel.SimpleChannel; +import org.apache.rocketmq.proxy.service.channel.SimpleChannelHandlerContext; +import org.apache.rocketmq.remoting.common.RemotingHelper; +import org.apache.rocketmq.remoting.netty.AttributeKeys; +import org.apache.rocketmq.remoting.protocol.LanguageCode; +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.header.EndTransactionRequestHeader; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class TransactionActivityTest extends InitConfigTest { + + private TransactionActivity transactionActivity; + @Mock + private MessagingProcessor messagingProcessor; + @Spy + private ChannelHandlerContext ctx = new SimpleChannelHandlerContext(new SimpleChannel(null, "0.0.0.0:0", "1.1.1.1:1")) { + @Override + public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) { + return null; + } + }; + + @Before + public void setUp() { + transactionActivity = new TransactionActivity(null, messagingProcessor); + Channel channel = ctx.channel(); + RemotingHelper.setPropertyToAttr(channel, AttributeKeys.CLIENT_ID_KEY, "clientId"); + RemotingHelper.setPropertyToAttr(channel, AttributeKeys.LANGUAGE_CODE_KEY, LanguageCode.JAVA); + RemotingHelper.setPropertyToAttr(channel, AttributeKeys.VERSION_KEY, MQVersion.CURRENT_VERSION); + } + + @Test + public void testEndTransactionWritesSuccessAfterFutureCompletes() throws Exception { + when(messagingProcessor.endTransaction(any(), eq("topic"), eq("transactionId"), eq("msgId"), + eq("producerGroup"), eq(TransactionStatus.COMMIT), eq(false))) + .thenReturn(CompletableFuture.completedFuture(null)); Review Comment: This test uses an already-completed future, so it doesn't verify the key regression requirement: that the success response is *not written* until after the future completes. Add a test that returns a pending `CompletableFuture`, asserts `writeAndFlush` has not been called immediately after `processRequest0(...)`, then completes the future and asserts `writeAndFlush` is called with `ResponseCode.SUCCESS`. -- 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]
