{"record":{"id":"01217c9f7e26a55a","repo":"alibaba/spring-cloud-alibaba","slug":"pull-consumer-already-running-this-tostring","errorCode":null,"errorMessage":"pull consumer already running. {this.toString()}","messagePattern":"pull consumer already running\\. (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"warning","filePath":"spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/integration/inbound/pull/RocketMQMessageSource.java","lineNumber":89,"sourceCode":"\tprivate volatile @Nullable Iterator<MessageExt> messageExtIterator = null;\n\n\tpublic RocketMQMessageSource(String name,\n\t\t\tExtendedConsumerProperties<RocketMQConsumerProperties> extendedConsumerProperties) {\n\t\tthis.topic = name;\n\t\tString subscription = extendedConsumerProperties.getExtension()\n\t\t\t\t.getSubscription();\n\t\tthis.messageSelector = RocketMQUtils.getMessageSelector(\n\t\t\t\tsubscription != null ? subscription : \"\");\n\t\tthis.extendedConsumerProperties = extendedConsumerProperties;\n\n\t}\n\n\t@Override\n\tpublic synchronized void start() {\n\t\tInstrumentation instrumentation = new Instrumentation(topic, this);\n\t\ttry {\n\t\t\tif (this.isRunning()) {\n\t\t\t\tthrow new IllegalStateException(\n\t\t\t\t\t\t\"pull consumer already running. \" + this.toString());\n\t\t\t}\n\t\t\tthis.consumer = RocketMQConsumerFactory\n\t\t\t\t\t.initPullConsumer(topic, extendedConsumerProperties);\n\t\t\t// This parameter must be 1, otherwise doReceive cannot be handled singly.\n\t\t\t// this.consumer.setPullBatchSize(1);\n\t\t\tthis.consumer.subscribe(topic, messageSelector);\n\t\t\tthis.consumer.setAutoCommit(false);\n\t\t\t// register TopicMessageQueueChangeListener for messageQueuesForTopic\n\t\t\tconsumer.registerTopicMessageQueueChangeListener(topic,\n\t\t\t\t\tmessageQueuesForTopic::put);\n\t\t\tthis.consumer.start();\n\t\t\t// Initialize messageQueuesForTopic immediately\n\t\t\tmessageQueuesForTopic.put(topic, consumer.fetchMessageQueues(topic));\n\t\t\tinstrumentation.markStartedSuccessfully();\n\t\t}\n\t\tcatch (MQClientException e) {\n\t\t\tinstrumentation.markStartFailed(e);","sourceCodeStart":71,"sourceCodeEnd":107,"githubUrl":"https://github.com/alibaba/spring-cloud-alibaba/blob/115d5901102009492e05d5ec18c3f79cad4077d0/spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/integration/inbound/pull/RocketMQMessageSource.java#L71-L107","documentation":"RocketMQMessageSource.start throws IllegalStateException if start() is invoked while the pull consumer is already running. The source tracks `running`; a second start without an intervening stop is a lifecycle misuse.","triggerScenarios":"start() is called again after a successful start with no intervening stop() - i.e. isRunning() is true.","commonSituations":"Duplicate bean wiring calling start twice; lifecycle restart logic not checking isRunning; an edge condition re-triggering start; manual start plus Spring-managed start.","solutions":["Ensure start() is called at most once between stop() calls.","If restarting is intended, call stop() first and confirm isRunning()==false.","Remove duplicate lifecycle triggers/wiring."],"exampleFix":"// before\nsource.start();\n// ...later, without stop()\nsource.start(); // -> [128]\n// after\nif (!source.isRunning()) {\n    source.start();\n}","handlingStrategy":"type-guard","validationCode":null,"typeGuard":"// Guard re-entrancy before start.\npublic synchronized void safeStart(RocketMQMessageSource source) {\n    if (source.isRunning()) {\n        return; // idempotent\n    }\n    source.start();\n}","tryCatchPattern":null,"preventionTips":["Make start() idempotent at the call site.","Always stop() before re-starting a pull consumer.","Audit lifecycle callers for duplicates."],"tags":["rocketmq","consumer","pull","lifecycle","programming-error"],"backgroundTag":null,"analyzedSha":"115d5901102009492e05d5ec18c3f79cad4077d0","analyzedAt":"2026-08-14T04:47:13.900Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}