:How to troubleshoot inconsistent subscription relationships in RocketMQ?

Commercial Java SDK

Maven coordinate example:

<groupId>com.aliyun.openservices</groupId>
<artifactId>ons-client</artifactId>
<version>x.x.x.Final</version>

Sample code:

Properties properties = new Properties();
// The Group ID that you created in the ApsaraMQ for RocketMQ console.
properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-xxxxx");
/*
* Some property configurations
*/
Consumer consumer = ONSFactory.createConsumer(properties);
// Subscribe to multiple tags.
consumer.subscribe("TopicTestMQ1", "TagA||TagB", new MessageListener() {...});
// Subscribe to multiple topics. Call subscribe again on the same consumer instance to add subscriptions.
consumer.subscribe("TopicTestMQ2", "TagA||TagB", new MessageListener() {...});

Community Java SDK (artifactId: rocketmq-client)

Maven coordinate example:

<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>x.x.x</version>

Sample code:

// The Group ID that you created in the ApsaraMQ for RocketMQ console.
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP_ID, getAclRPCHook(), new AllocateMessageQueueAveragely(), true, null);
/*
* Some property configurations
*/
// Subscribe to multiple tags
consumer.subscribe("java-test", "TagA||TagB");
// Subscribe to a second topic. Note that a consumer can register only one consumption logic. This means the same logic applies to all subscriptions.
consumer.subscribe("cxtest", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {...});

Community Java SDK (artifactId: rocketmq-client-java)

Maven coordinate example:

<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client-java</artifactId>
<version>5.0.8</version>

Sample code:

final ClientServiceProvider provider = ClientServiceProvider.loadService();
/*
* Some configurations
*/
HashMap<String, FilterExpression> sub = new HashMap<>();
// Subscribe to multiple tags
sub.put("Topic1", new FilterExpression("TagA||TagB", FilterExpressionType.TAG));
// Subscribe to a second topic. Note that a consumer can register only one consumption logic. This means the same logic applies to all subscriptions.
sub.put("Topic2", new FilterExpression("TagA||TagB", FilterExpressionType.TAG));
PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // Set the consumer group. Use the Group ID created in the ApsaraMQ for RocketMQ console.
                .setConsumerGroup(consumerGroup)
                // Set the pre-attached subscription relationships.
                .setSubscriptionExpressions(sub)
                // Set the message listener.
                .setMessageListener(messageView -> {...})
                .build();

SpringBoot framework (using the RocketMQMessageListener annotation)

If you use the RocketMQMessageListener annotation to start a consumer, note that each annotation starts a separate consumer instance. If you use multiple annotations in your code and each subscribes to a different topic, this causes an inconsistent subscription relationship.

For example, the following code implements the RocketMQListener interface twice, which subscribes to TopicA and TopicB in the same group. At runtime, the client creates two consumer instances. This results in two consumers in the same consumer group subscribing to TopicA and TopicB respectively, which causes an inconsistent subscription relationship.

// Business logic 1: TestMessageListener1.java
@Component
@RocketMQMessageListener(consumerGroup="GID_A", topic = "TopicA")
public class TestMessageListener1 implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt messageExt) {
        /*
        * Some configurations
        */
    }
}
// Business logic 2: TestMessageListener2.java
@Component
@RocketMQMessageListener(consumerGroup="GID_A", topic = "TopicB")
public class TestMessageListener2 implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt messageExt) {
        /*
        * Some configurations
        */
    }
}

To implement multiple subscriptions, you can rewrite the prepareStart method of the RocketMQPushConsumerLifecycleListener interface. The following code provides an example:

@Component
@RocketMQMessageListener(consumerGroup = "GID_A", topic = "")
public class TestMessageListener implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    @Override
    public void onMessage(MessageExt messageExt) {}
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        try{
            // Subscribe to multiple tags
            defaultMQPushConsumer.subscribe("TopicA", "TagA||TagB");
            // Subscribe to a second topic
            defaultMQPushConsumer.subscribe("TopicB", "*");
            // Register the consumption logic
            defaultMQPushConsumer.registerMessageListener((List<MessageExt> messages, ConsumeConcurrentlyContext context) -> {......});
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

SpringBoot framework (using the Bean annotation)

This method manually injects a ConsumerBean. You must use the native SDK method to create multiple subscriptions.

@Configuration
public class ConsumerClient {
  
    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ConsumerBean buildConsumer() {
        // Inject the corresponding ConsumerBean here. Refer to the multiple subscription methods for the commercial and community SDKs.
    }
}