[Bug] RocketMQ consumer消费不到第一条消息

[zooah212]

Before Creating the Bug Report

• I found a bug, not just asking a question, which should be created in GitHub Discussions.

• I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.

• I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Runtime platform environment

CentOS Linux 7

RocketMQ version

5.1.3

JDK Version

JDK 8

Describe the Bug

全新的RocketMQ环境，里面没有Consumer和Topic。StringBoot项目集群RocketMQ5.1.3版本，首先启动Consumer，此时查看dashboard，能查到Consumer，Consumer要消费的Topic还没有创建；然后启动Producer，并发送一条消息，此时收不到第一条消息，然后Producer再发一条消息，Consumer能消费到消息；
ConsumeFromWhere：Last_Offset
消费模式：广播模式和集群模式都会由此现象
如果先发消息，再启动Consumer，广播消费模式会消费到之前的数据；集群消费模式，虽然在dashboard中看到消息的状态是已消费，但是在程序中Consumer还是没有收到消息，不知道消息被谁消费了；

Steps to Reproduce

看bug描述

What Did You Expect to See?

即使是全新的环境，先启动consumer，也能正常消费到数据。

What Did You See Instead?

无

Additional Context

无

[SchopenhauerZhang]

我遇到过，不过我是5.0版本。
我理解是第一条消息发送时由于topic没有创建，所以第一条消息去触发了topic的创建（其实是consume queue的初始化）然后由于topic还没有创建（在消息被broker接收到时）所以消息发送失败了（write consume queue fail），所以在这里应该加一个重试，我原本打算debug一下这块的细节，不过还没来得及做。
不过丢失了第一条消息确实是一个已知的问题。

[zooah212]

我遇到过，不过我是5.0版本。 我理解是第一条消息发送时由于topic没有创建，所以第一条消息去触发了topic的创建（其实是consume queue的初始化）然后由于topic还没有创建（在消息被broker接收到时）所以消息发送失败了（write consume queue fail），所以在这里应该加一个重试，我原本打算debug一下这块的细节，不过还没来得及做。 不过丢失了第一条消息确实是一个已知的问题。

感谢，我测试了 5.0.0 和 5.1.3 都是有这个问题。而且如果是 mq 集群时，这个问题更让人头疼，单机时至少发第二次能够发出去，但是集群时如果正好创建了 topic 的 broker 宕机，那么后续再发消息其他的 broker 不会再自动创建 topic，除非重启应用。

[mufengCc]

还真的是。我也复现这个问题了。 同样的5.1.3版本

[zooah212]

还真的是。我也复现这个问题了。 同样的5.1.3版本

嗯。当在高可用场景下，会出现更严重的问题。如果有两个 master A 和 B，当发消息的时候会随机在某个 master 上面创建 topic，比如在 master A 上面创建好了 topic，之后发/收消息会正常；但是当 master A 挂掉之后，再发消息则 Producer 会直接报错，发不出消息，且 master B 上面也不会再创建 topic，除非重启应用。。。

[SchopenhauerZhang]

收到，感谢提问。我正在本地复现，并且正在尝试基于5.0版本修复这个问题。按照以上大家的描述我理解是一个共性问题，所以等我PR，不过由于工作的原因，可能需要点时间，烦请暂时克服一下，我尽量加快修复速度。:-）

[xinchilde]

收到，感谢提问。我正在本地复现，并且正在尝试基于5.0版本修复这个问题。按照以上大家的描述我理解是一个共性问题，所以等我PR，不过由于工作的原因，可能需要点时间，烦请暂时克服一下，我尽量加快修复速度。:-）

您好，这个问题解决了么？希望能百忙之中抽空处理下，或通知其他贡献者研究下。谢谢了~

[wolf27w]

我的是4.9.7版本也出现这个问题，消费者无法消费第一条数据

[qinyi8888]

目前自己创建,临时解决方法：
引入配置文件：
implementation('org.apache.rocketmq:rocketmq-tools:5.0.0')
简单实现代码：
`/**

• @author

• @Date 2024/12/20
  */
  @slf4j
  @component
  public class InitRocketmqTopicRunner implements ApplicationRunner {

  @resource
  private RedisService redisService;

  @value("${rocketmq.name-server}")
  private String nameServerAddr;

  @OverRide
  public void run(ApplicationArguments args) throws Exception {
  Object flag = redisService.get(RedisKeys.SYSTEM_INITIALIZATION_FLAG);
  if (flag == null) {
  log.info("Init Rocketmq Topic");
  initTopic();
  redisService.setCacheObject(RedisKeys.SYSTEM_INITIALIZATION_FLAG, true);
  }
  }

  private void initTopic() {
  DefaultMQAdminExt admin = new DefaultMQAdminExt();
  admin.setInstanceName(Long.toString(System.currentTimeMillis()));
  admin.setNamesrvAddr(nameServerAddr);
  try {
  admin.start();
  ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
  Set clusterSet = clusterInfo.getClusterAddrTable().keySet();
  if (clusterSet.isEmpty()) {
  log.error("No clusters found in RocketMQ.");
  return;
  }
  // Use the first cluster found
  String clusterName = clusterSet.iterator().next();
  log.info("Using cluster: {}", clusterName);
  Field[] fields = TopicConstants.class.getDeclaredFields();
  for (Field field : fields) {
  if (field.getType() == String.class) {
  String topic = (String) field.get(null);
  if (checkIfTopicExists(admin, topic)) {
  log.info("Topic {} already exists.", topic);
  } else {
  createTopic(admin,clusterName, topic);
  }
  }
  }
  } catch (Exception e) {
  log.error("Failed to initialize RocketMQ topics", e);
  } finally {
  admin.shutdown();
  }
  }

  private boolean checkIfTopicExists(DefaultMQAdminExt admin, String topic) {
  try {
  TopicRouteData topicRouteData = admin.examineTopicRouteInfo(topic);
  return topicRouteData != null && !topicRouteData.getQueueDatas().isEmpty();
  } catch (MQClientException | RemotingException | InterruptedException e) {
  // If the topic does not exist, an exception will be thrown
  log.warn("Topic {} does not exist.", topic);
  return false;
  }
  }
  private void createTopic(DefaultMQAdminExt admin, String clusterName, String topic) throws Exception{
  admin.createTopic(clusterName, topic, 4,null);
  log.info("Created topic {}", topic);
  }`
  'public interface TopicConstants {
  String DOWN_INVOKE_SERVICE = "down_invoke_service";
  }'

[lehug]

5.2.0版本，遇到类似的问题，producer发送成功，根据id通过mqadmin也查不到，后续跟踪到store/index文件里面缺失信息

[mengnankkkk]

大家好，这个问题是一个时序问题，一个topic被创建的时候，队列为null,最大偏移量为0，然后生产者发送第一条消息的时候，最大偏移量为1。消费者启动的时候，查询最大偏移量为-1，然后调用maxoffset(mq),就会返回1.这样0消息就被跳过了。
可以在发送消息前，可以先启动 Consumer 实例，或者在 Consumer 启动后等待一段时间再发送消息。
然后在配置的时候，设置从ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET开始消费。
也可以在代码方面更改，#9552