[ROCKETMQ-239]fix query message by time and fix get queue offset by time

broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java

@@ -20,12 +20,6 @@
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.FileRegion;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentMap;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.client.ClientChannelInfo;
 import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
@@ -34,8 +28,7 @@
 import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.common.constant.OffsetConstant;
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.common.message.MessageQueueForC;
 import org.apache.rocketmq.common.protocol.RequestCode;
@@ -47,12 +40,21 @@
 import org.apache.rocketmq.common.protocol.header.GetConsumerStatusRequestHeader;
 import org.apache.rocketmq.common.protocol.header.NotifyConsumerIdsChangedRequestHeader;
 import org.apache.rocketmq.common.protocol.header.ResetOffsetRequestHeader;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
 import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
 import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.store.SelectMappedBufferResult;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentMap;
+
 public class Broker2Client {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
     private final BrokerController brokerController;
@@ -151,7 +153,8 @@ public RemotingCommand resetOffset(String topic, String group, long timeStamp, b
 
                 timeStampOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
             } else {
-                timeStampOffset = this.brokerController.getMessageStore().getOffsetInQueueByTime(topic, i, timeStamp);
+                timeStampOffset = this.brokerController.getMessageStore().getOffsetInQueueByTime(topic, i, timeStamp,
+                    OffsetConstant.SEARCH_OFFSET_BYTIME_RETURN_RETURN_FIRST_OFFSET);
             }
 
             if (timeStampOffset < 0) {

broker/src/main/java/org/apache/rocketmq/broker/plugin/AbstractPluginMessageStore.java

@@ -107,8 +107,8 @@ public long getCommitLogOffsetInQueue(String topic, int queueId, long consumeQue
     }
 
     @Override
-    public long getOffsetInQueueByTime(String topic, int queueId, long timestamp) {
-        return next.getOffsetInQueueByTime(topic, queueId, timestamp);
+    public long getOffsetInQueueByTime(String topic, int queueId, long timestamp, int getLastOrFirstOffset) {
+        return next.getOffsetInQueueByTime(topic, queueId, timestamp, getLastOrFirstOffset);
     }
 
     @Override

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

@@ -44,6 +44,7 @@
 import org.apache.rocketmq.common.admin.TopicOffset;
 import org.apache.rocketmq.common.admin.TopicStatsTable;
 import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.constant.OffsetConstant;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.common.message.MessageDecoder;
@@ -364,8 +365,13 @@ private RemotingCommand searchOffsetByTimestamp(ChannelHandlerContext ctx,
         final SearchOffsetRequestHeader requestHeader =
             (SearchOffsetRequestHeader) request.decodeCommandCustomHeader(SearchOffsetRequestHeader.class);
 
+        int getLastOrFirstOffset = OffsetConstant.SEARCH_OFFSET_BYTIME_RETURN_RETURN_FIRST_OFFSET;
+        if (requestHeader.getGetLastOrFirstOffset() != null) {
+            getLastOrFirstOffset = requestHeader.getGetLastOrFirstOffset();
+        }
+
         long offset = this.brokerController.getMessageStore().getOffsetInQueueByTime(requestHeader.getTopic(), requestHeader.getQueueId(),
-            requestHeader.getTimestamp());
+            requestHeader.getTimestamp(), getLastOrFirstOffset);
 
         responseHeader.setOffset(offset);
 

client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java

@@ -117,6 +117,11 @@ public long searchOffset(MessageQueue mq, long timestamp) throws MQClientExcepti
         return this.defaultMQPullConsumerImpl.searchOffset(mq, timestamp);
     }
 
+    @Override
+    public long searchOffset(MessageQueue mq, long timestamp, int getLastOrFirstOffset) throws MQClientException {
+        return this.defaultMQPullConsumerImpl.searchOffset(mq, timestamp, getLastOrFirstOffset);
+    }
+
     @Override
     public long maxOffset(MessageQueue mq) throws MQClientException {
         return this.defaultMQPullConsumerImpl.maxOffset(mq);

client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java

@@ -300,6 +300,11 @@ public long searchOffset(MessageQueue mq, long timestamp) throws MQClientExcepti
         return this.defaultMQPushConsumerImpl.searchOffset(mq, timestamp);
     }
 
+    @Override
+    public long searchOffset(MessageQueue mq, long timestamp, int getLastOrFirstOffset) throws MQClientException {
+        return this.defaultMQPushConsumerImpl.searchOffset(mq, timestamp, getLastOrFirstOffset);
+    }
+
     @Override
     public long maxOffset(MessageQueue mq) throws MQClientException {
         return this.defaultMQPushConsumerImpl.maxOffset(mq);

client/src/main/java/org/apache/rocketmq/client/consumer/MQConsumer.java

@@ -48,4 +48,6 @@ void sendMessageBack(final MessageExt msg, final int delayLevel, final String br
      * @return queue set
      */
     Set<MessageQueue> fetchSubscribeMessageQueues(final String topic) throws MQClientException;
+
+    long searchOffset(final MessageQueue mq, final long timestamp, final int getLastOrFirstOffset) throws MQClientException;
 }

client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java

@@ -34,6 +34,7 @@
 import org.apache.rocketmq.client.log.ClientLogger;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.constant.OffsetConstant;
 import org.apache.rocketmq.common.help.FAQUrl;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.common.message.MessageClientIDSetter;
@@ -167,6 +168,10 @@ public Set<MessageQueue> fetchSubscribeMessageQueues(String topic) throws MQClie
     }
 
     public long searchOffset(MessageQueue mq, long timestamp) throws MQClientException {
+        return this.doSearchOffset(mq, timestamp, OffsetConstant.SEARCH_OFFSET_BYTIME_RETURN_RETURN_FIRST_OFFSET);
+    }
+
+    public long doSearchOffset(MessageQueue mq, long timestamp, int getLastOrFirstOffset) throws MQClientException {
         String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
         if (null == brokerAddr) {
             this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
@@ -176,7 +181,7 @@ public long searchOffset(MessageQueue mq, long timestamp) throws MQClientExcepti
         if (brokerAddr != null) {
             try {
                 return this.mQClientFactory.getMQClientAPIImpl().searchOffset(brokerAddr, mq.getTopic(), mq.getQueueId(), timestamp,
-                    timeoutMillis);
+                    getLastOrFirstOffset, timeoutMillis);
             } catch (Exception e) {
                 throw new MQClientException("Invoke Broker[" + brokerAddr + "] exception", e);
             }

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -668,14 +668,14 @@ public MessageExt viewMessage(final String addr, final long phyoffset, final lon
 
         throw new MQBrokerException(response.getCode(), response.getRemark());
     }
-
-    public long searchOffset(final String addr, final String topic, final int queueId, final long timestamp,
-        final long timeoutMillis)
+
+    public long searchOffset(final String addr, final String topic, final int queueId, final long timestamp, final int getLastOrFirstOffset, final long timeoutMillis)
         throws RemotingException, MQBrokerException, InterruptedException {
         SearchOffsetRequestHeader requestHeader = new SearchOffsetRequestHeader();
         requestHeader.setTopic(topic);
         requestHeader.setQueueId(queueId);
         requestHeader.setTimestamp(timestamp);
+        requestHeader.setGetLastOrFirstOffset(getLastOrFirstOffset);
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.SEARCH_OFFSET_BY_TIMESTAMP, requestHeader);
 
         RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java

@@ -16,13 +16,6 @@
  */
 package org.apache.rocketmq.client.impl.consumer;
 
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.ConcurrentMap;
 import org.apache.rocketmq.client.QueryResult;
 import org.apache.rocketmq.client.Validators;
 import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
@@ -45,10 +38,10 @@
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.ServiceState;
 import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.constant.OffsetConstant;
 import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
 import org.apache.rocketmq.common.filter.FilterAPI;
 import org.apache.rocketmq.common.help.FAQUrl;
-import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageAccessor;
 import org.apache.rocketmq.common.message.MessageConst;
@@ -59,10 +52,19 @@
 import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
 import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
 import org.apache.rocketmq.common.sysflag.PullSysFlag;
+import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
 import org.apache.rocketmq.remoting.exception.RemotingException;
 
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+
 public class DefaultMQPullConsumerImpl implements MQConsumerInner {
     private final InternalLogger log = ClientLogger.getLog();
     private final DefaultMQPullConsumer defaultMQPullConsumer;
@@ -471,8 +473,12 @@ public MessageExt queryMessageByUniqKey(String topic, String uniqKey)
     }
 
     public long searchOffset(MessageQueue mq, long timestamp) throws MQClientException {
+        return this.searchOffset(mq, timestamp, OffsetConstant.SEARCH_OFFSET_BYTIME_RETURN_RETURN_FIRST_OFFSET);
+    }
+
+    public long searchOffset(final MessageQueue mq, final long timestamp, final int getLastOrFirstOffset) throws MQClientException {
         this.makeSureStateOK();
-        return this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
+        return this.mQClientFactory.getMQAdminImpl().doSearchOffset(mq, timestamp, getLastOrFirstOffset);
     }
 
     public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName)

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java

@@ -53,6 +53,7 @@
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.ServiceState;
 import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.constant.OffsetConstant;
 import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
 import org.apache.rocketmq.common.filter.FilterAPI;
 import org.apache.rocketmq.common.help.FAQUrl;
@@ -956,7 +957,11 @@ public void resetOffsetByTimeStamp(long timeStamp)
     }
 
     public long searchOffset(MessageQueue mq, long timestamp) throws MQClientException {
-        return this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
+        return this.searchOffset(mq, timestamp, OffsetConstant.SEARCH_OFFSET_BYTIME_RETURN_RETURN_FIRST_OFFSET);
+    }
+
+    public long searchOffset(MessageQueue mq, long timestamp, int getLastOrFirstOffset) throws MQClientException {
+        return this.mQClientFactory.getMQAdminImpl().doSearchOffset(mq, timestamp, getLastOrFirstOffset);
     }
 
     @Override

common/src/main/java/org/apache/rocketmq/common/constant/OffsetConstant.java

@@ -0,0 +1,26 @@
+/*
+ * 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.common.constant;
+
+public class OffsetConstant {
+
+    public static final int SEARCH_OFFSET_BYTIME_RETURN_RETURN_FIRST_OFFSET = 1;
+
+    public static final int SEARCH_OFFSET_BYTIME_RETURN_RETURN_LAST_OFFSET = 2;
+
+}

common/src/main/java/org/apache/rocketmq/common/protocol/header/SearchOffsetRequestHeader.java

@@ -32,6 +32,8 @@ public class SearchOffsetRequestHeader implements CommandCustomHeader {
     @CFNotNull
     private Long timestamp;
 
+    private Integer getLastOrFirstOffset;
+
     @Override
     public void checkFields() throws RemotingCommandException {
 
@@ -61,4 +63,11 @@ public void setTimestamp(Long timestamp) {
         this.timestamp = timestamp;
     }
 
+    public Integer getGetLastOrFirstOffset() {
+        return getLastOrFirstOffset;
+    }
+
+    public void setGetLastOrFirstOffset(Integer getLastOrFirstOffset) {
+        this.getLastOrFirstOffset = getLastOrFirstOffset;
+    }
 }