diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommand.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommand.java index 25066ea5305..1d7b10e2cea 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommand.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommand.java @@ -16,6 +16,7 @@ */ package org.apache.rocketmq.proxy.service.message; +import com.google.common.base.Stopwatch; import java.util.HashMap; import org.apache.rocketmq.remoting.CommandCustomHeader; import org.apache.rocketmq.remoting.protocol.LanguageCode; @@ -31,6 +32,8 @@ public static LocalRemotingCommand createRequestCommand(int code, CommandCustomH cmd.setExtFields(new HashMap<>()); setCmdVersion(cmd); cmd.makeCustomHeaderToNet(); + // Local requests bypass NettyDecoder, so initialize the processing timer explicitly. + cmd.setProcessTimer(Stopwatch.createStarted()); return cmd; } } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommandTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommandTest.java new file mode 100644 index 00000000000..121f8d5e44f --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommandTest.java @@ -0,0 +1,80 @@ +/* + * 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.service.message; + +import io.netty.channel.ChannelFuture; +import io.netty.channel.DefaultChannelPromise; +import io.netty.util.concurrent.ImmediateEventExecutor; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongHistogram; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.rocketmq.common.metrics.NopLongHistogram; +import org.apache.rocketmq.proxy.service.channel.SimpleChannel; +import org.apache.rocketmq.remoting.metrics.RemotingMetricsManager; +import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.apache.rocketmq.remoting.protocol.RequestCode; +import org.junit.Test; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +public class LocalRemotingCommandTest { + + @Test + public void testCreateRequestCommandShouldStartProcessTimer() { + LocalRemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, null, "JAVA"); + + assertNotNull(command.getProcessTimer()); + assertTrue(command.getProcessTimer().isRunning()); + } + + @Test + public void testWriteResponseShouldRecordLatencyForLocalRemotingCommand() { + LocalRemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, null, "JAVA"); + RemotingCommand response = RemotingCommand.createResponseCommand(0, null); + AtomicBoolean latencyRecorded = new AtomicBoolean(false); + LongHistogram rpcLatency = new NopLongHistogram() { + @Override + public void record(long value, Attributes attributes) { + latencyRecorded.set(true); + } + }; + RemotingMetricsManager remotingMetricsManager = new RemotingMetricsManager() { + @Override + public LongHistogram getRpcLatency() { + return rpcLatency; + } + }; + SimpleChannel channel = new SimpleChannel("127.0.0.1:8080", "127.0.0.1:8081") { + @Override + public ChannelFuture writeAndFlush(Object msg) { + return new DefaultChannelPromise(this, ImmediateEventExecutor.INSTANCE).setSuccess(); + } + }; + + NettyRemotingAbstract.writeResponse( + channel, + command, + response, + null, + remotingMetricsManager + ); + + assertTrue(latencyRecorded.get()); + } +}