How to use sql92 in rocketMQ to filter messages !maybe not a bug

weixiaoming369 commented 4 weeks ago

Before Creating the Bug Report

[X] I found a bug, not just asking a question, which should be created in GitHub Discussions.
[X] I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.
[X] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Runtime platform environment

win 11

RocketMQ version

4.9.4

JDK Version

1.8

Describe the Bug

I'm testing rocketMQ to enable sql92 to filter messages, but according to the official documentation steps and online references, I can't use sql92 to filter messages normally, it doesn't always report errors, but the frequency of errors is basically 100%

Steps to Reproduce

1、my pom

org.apache.rocketmq rocketmq-client 4.9.4

2、my rocketMQ window zip rocketmq-all-4.9.4 3、broker.conf and broker-a.properties and broker-a-s.properties Both have added Inable Propelt = true 4、this is my producer test code

5、my consumer test code /**

顺序消费

@throws Exception */ public static void consumer_order() throws Exception {

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sql_consumer");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.subscribe("test", MessageSelector.bySql("(TAGS IS NOT NULL AND TAGS IN ('Earth', 'Mars'))"
        + "and country = 'China'"));
consumer.setMessageModel(MessageModel.CLUSTERING);
consumer.registerMessageListener(new MessageListenerOrderly() {
    AtomicLong consumerTimes = new AtomicLong(0);
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext consumeOrderlyContext) {
        for (MessageExt msg : msgs) {
            this.consumerTimes.incrementAndGet();
            try {
                System.err.printf("%s receiveMessage : %s%n", Thread.currentThread().getName(),
                        new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET));
            } catch (UnsupportedEncodingException e) {
                throw new RuntimeException(e);
            }
            if (this.consumerTimes.get() % 2 == 0) {
                return ConsumeOrderlyStatus.SUCCESS;
            } else if (this.consumerTimes.get() % 5 == 0) {
                consumeOrderlyContext.setSuspendCurrentQueueTimeMillis(3000);
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
            return ConsumeOrderlyStatus.SUCCESS;
        }
        return null;
    }
});
consumer.start();
System.out.printf("consumer started %n");

} 6、this is err info messages when boot test

What Did You Expect to See?

11111111111111111111

What Did You See Instead?

1111111111111111

Additional Context

yx9o commented 4 weeks ago

Hi, can you confirm whether the enablePropertyFilter property value is loaded correctly?

weixiaoming369 commented 3 weeks ago

Thanks for the reply, I don't know how to see if the enablePropertyFilter is loaded correctly, I don't have any clue other than adding this property to the config file and giving it a value of true I've observed that the SQL filtering code I wrote is documented in the filter.log, which may indicate that enablePropertyFilter is already loaded

weixiaoming369 commented 3 weeks ago

Maybe I might have figured out the problem, but now I'm confused that when I start the broker with mqadmin mqbroker.cmd -n localhost:9876, it doesn't seem to be using the default configuration file broker.conf This can be deduced from the log information recorded by the broker.log, since I don't set any value for brokerName, and the value of the log record is inconsistent with the default configuration file, so from here I observe the corresponding enablePropertyFilter=false When I specified the profile, it became working this is my test result So I'm wondering why the default config file isn't used, and maybe using a program instead of the command line doesn't give up this issue, but I'll have to wait until I actually test it

apache / rocketmq