{"record":{"id":"e54c701007bd5b4c","repo":"apache/rocketmq","slug":"messagelistener-must-be-instanceof-messagelistener","errorCode":null,"errorMessage":"messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently","messagePattern":"messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently","errorType":"validation","errorClass":"MQClientException","httpStatus":null,"severity":"error","filePath":"client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java","lineNumber":1093,"sourceCode":"        if (null == this.defaultMQPushConsumer.getSubscription()) {\n            throw new MQClientException(\n                \"subscription is null\"\n                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),\n                null);\n        }\n\n        // messageListener\n        if (null == this.defaultMQPushConsumer.getMessageListener()) {\n            throw new MQClientException(\n                \"messageListener is null\"\n                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),\n                null);\n        }\n\n        boolean orderly = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerOrderly;\n        boolean concurrently = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerConcurrently;\n        if (!orderly && !concurrently) {\n            throw new MQClientException(\n                \"messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently\"\n                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),\n                null);\n        }\n\n        // consumeThreadMin\n        if (this.defaultMQPushConsumer.getConsumeThreadMin() < 1\n            || this.defaultMQPushConsumer.getConsumeThreadMin() > 1000) {\n            throw new MQClientException(\n                \"consumeThreadMin Out of range [1, 1000]\"\n                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),\n                null);\n        }\n\n        // consumeThreadMax\n        if (this.defaultMQPushConsumer.getConsumeThreadMax() < 1 || this.defaultMQPushConsumer.getConsumeThreadMax() > 1000) {\n            throw new MQClientException(\n                \"consumeThreadMax Out of range [1, 1000]\"","sourceCodeStart":1075,"sourceCodeEnd":1111,"githubUrl":"https://github.com/apache/rocketmq/blob/293f5885719fc4aa3619446a1900f58ccfcfdd29/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java#L1075-L1111","documentation":"checkConfig() accepts only MessageListenerOrderly or MessageListenerConcurrently as the listener type (checked via instanceof after the null check). Any other MessageListener implementation (the base interface has others, e.g. MessageListenerOrderly/Concurrently are the supported concrete contracts) is rejected because the push consumer's dispatch machinery only implements these two protocols.","triggerScenarios":"Passing a listener that implements only the base org.apache.rocketmq.client.consumer.listener.MessageListener interface; passing a custom subinterface; passing a lambdas targeted at an unsupported functional interface.","commonSituations":"Implementing the generic interface 'to be flexible'; upgrading from old client versions where listener handling differed; mocking MessageListener in tests.","solutions":["Implement MessageListenerConcurrently (default choice) or MessageListenerOrderly (for ordered consumption with queue-level locking)","If you wrote a custom interface, wrap it in an adapter class implementing one of the two supported listeners","Use the exact lambdas: (MessageListenerConcurrently) or (MessageListenerOrderly) casts so the compiler targets the right interface"],"exampleFix":"// before\nclass MyListener implements MessageListener { ... } // base interface only\nc.registerMessageListener(new MyListener()); // rejected\n\n// after\nclass MyListener implements MessageListenerConcurrently {\n    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext ctx) { ... return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }\n}\nc.registerMessageListener(new MyListener());","handlingStrategy":"type-guard","validationCode":"Objects.requireNonNull(listener, \"listener required\");\nif (!(listener instanceof MessageListenerConcurrently) && !(listener instanceof MessageListenerOrderly))\n    throw new IllegalArgumentException(\"Listener must be MessageListenerConcurrently or MessageListenerOrderly\");\nconsumer.registerMessageListener(listener);","typeGuard":"boolean isSupportedListener(MessageListener l) {\n    return l instanceof MessageListenerConcurrently || l instanceof MessageListenerOrderly;\n}","tryCatchPattern":null,"preventionTips":["Implement one of the two supported interfaces; never the bare MessageListener","When using lambdas, cast explicitly: (MessageListenerConcurrently) (msgs, ctx) -> ...","Wrap custom listener logic in an adapter of the supported types"],"tags":["rocketmq","config","listener","type-check"],"backgroundTag":null,"analyzedSha":"293f5885719fc4aa3619446a1900f58ccfcfdd29","analyzedAt":"2026-08-14T11:50:13.822Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}