alibaba/spring-cloud-alibaba · warning · IllegalStateException

pull consumer already running. {this.toString()}

Error message

pull consumer already running. {this.toString()}

What it means

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.

Source

Thrown at spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/integration/inbound/pull/RocketMQMessageSource.java:89

	private volatile @Nullable Iterator<MessageExt> messageExtIterator = null;

	public RocketMQMessageSource(String name,
			ExtendedConsumerProperties<RocketMQConsumerProperties> extendedConsumerProperties) {
		this.topic = name;
		String subscription = extendedConsumerProperties.getExtension()
				.getSubscription();
		this.messageSelector = RocketMQUtils.getMessageSelector(
				subscription != null ? subscription : "");
		this.extendedConsumerProperties = extendedConsumerProperties;

	}

	@Override
	public synchronized void start() {
		Instrumentation instrumentation = new Instrumentation(topic, this);
		try {
			if (this.isRunning()) {
				throw new IllegalStateException(
						"pull consumer already running. " + this.toString());
			}
			this.consumer = RocketMQConsumerFactory
					.initPullConsumer(topic, extendedConsumerProperties);
			// This parameter must be 1, otherwise doReceive cannot be handled singly.
			// this.consumer.setPullBatchSize(1);
			this.consumer.subscribe(topic, messageSelector);
			this.consumer.setAutoCommit(false);
			// register TopicMessageQueueChangeListener for messageQueuesForTopic
			consumer.registerTopicMessageQueueChangeListener(topic,
					messageQueuesForTopic::put);
			this.consumer.start();
			// Initialize messageQueuesForTopic immediately
			messageQueuesForTopic.put(topic, consumer.fetchMessageQueues(topic));
			instrumentation.markStartedSuccessfully();
		}
		catch (MQClientException e) {
			instrumentation.markStartFailed(e);

View on GitHub (pinned to 115d590110)

Solutions

1. Ensure start() is called at most once between stop() calls.
2. If restarting is intended, call stop() first and confirm isRunning()==false.
3. Remove duplicate lifecycle triggers/wiring.

Example fix

// before
source.start();
// ...later, without stop()
source.start(); // -> [128]
// after
if (!source.isRunning()) {
    source.start();
}

// Guard re-entrancy before start.
public synchronized void safeStart(RocketMQMessageSource source) {
    if (source.isRunning()) {
        return; // idempotent
    }
    source.start();
}

When it happens

Trigger: start() is called again after a successful start with no intervening stop() - i.e. isRunning() is true.

Common situations: Duplicate bean wiring calling start twice; lifecycle restart logic not checking isRunning; an edge condition re-triggering start; manual start plus Spring-managed start.

Related errors

• Already acknowledged
• group must be configured for DLQ{topic}
• DefaultMQPushConsumer not initialized
• The message queue is not in assigned list
• group must be configured for DLQ{destination.getName()}

AI-assisted analysis of alibaba/spring-cloud-alibaba@115d590110 (2026-08-14). Data as JSON: /api/errors/01217c9f7e26a55a. Report an issue: GitHub.