[ISSUE #9862] [Bug] Fix the issue of missing bornTime in POP requests when broker ACL is enabled (#9863)

Author: dingshuangxi888

broker/src/main/java/org/apache/rocketmq/broker/processor/NotificationProcessor.java

@@ -19,7 +19,6 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Random;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.longpolling.PollingHeader;
@@ -79,16 +78,17 @@ public void notifyMessageArriving(final String topic, final int queueId) {
     @Override
     public RemotingCommand processRequest(final ChannelHandlerContext ctx,
         RemotingCommand request) throws RemotingCommandException {
-        request.addExtFieldIfNotExist(BORN_TIME, String.valueOf(System.currentTimeMillis()));
-        if (Objects.equals(request.getExtFields().get(BORN_TIME), "0")) {
-            request.addExtField(BORN_TIME, String.valueOf(System.currentTimeMillis()));
-        }
         Channel channel = ctx.channel();
 
         RemotingCommand response = RemotingCommand.createResponseCommand(NotificationResponseHeader.class);
         final NotificationResponseHeader responseHeader = (NotificationResponseHeader) response.readCustomHeader();
         final NotificationRequestHeader requestHeader =
-            (NotificationRequestHeader) request.decodeCommandCustomHeader(NotificationRequestHeader.class);
+            request.decodeCommandCustomHeader(NotificationRequestHeader.class, true);
+        if (requestHeader.getBornTime() == 0) {
+            final long beginTimeMills = this.brokerController.getMessageStore().now();
+            request.addExtField(BORN_TIME, String.valueOf(beginTimeMills));
+            requestHeader.setBornTime(beginTimeMills);
+        }
 
         response.setOpaque(request.getOpaque());
 

broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java

@@ -29,7 +29,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Objects;
 import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -228,18 +227,16 @@ public RemotingCommand processRequest(final ChannelHandlerContext ctx, RemotingC
 
         final long beginTimeMills = this.brokerController.getMessageStore().now();
 
-        // fill bron time to properties if not exist, why we need this?
-        request.addExtFieldIfNotExist(BORN_TIME, String.valueOf(System.currentTimeMillis()));
-        if (Objects.equals(request.getExtFields().get(BORN_TIME), "0")) {
-            request.addExtField(BORN_TIME, String.valueOf(System.currentTimeMillis()));
-        }
-
         Channel channel = ctx.channel();
         RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
         response.setOpaque(request.getOpaque());
 
         final PopMessageRequestHeader requestHeader =
             request.decodeCommandCustomHeader(PopMessageRequestHeader.class, true);
+        if (requestHeader.getBornTime() == 0) {
+            request.addExtField(BORN_TIME, String.valueOf(beginTimeMills));
+            requestHeader.setBornTime(beginTimeMills);
+        }
         final PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
 
         // Pop mode only supports consumption in cluster load balancing mode

proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java

@@ -199,7 +199,6 @@ public CompletableFuture<Void> endTransactionOneway(ProxyContext ctx, String bro
     @Override
     public CompletableFuture<PopResult> popMessage(ProxyContext ctx, AddressableMessageQueue messageQueue,
         PopMessageRequestHeader requestHeader, long timeoutMillis) {
-        requestHeader.setBornTime(System.currentTimeMillis());
         RemotingCommand request = LocalRemotingCommand.createRequestCommand(RequestCode.POP_MESSAGE, requestHeader, ctx.getLanguage());
         CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
         SimpleChannel channel = channelManager.createInvocationChannel(ctx);