apache / apache/rocketmq

[Enhancement] Support `MessageQueueSelector` for transactional messages

Open
#10,545 2 comments 0 reactions 0 assignees View on GitHub
type/enhancement
Dominant language
Java
Stars
22.6k
Forks
12k
Avg merge
3d 1h
Merged PRs (30d)
27

Description

### Before Creating the Enhancement Request

- [x] I have confirmed that this should be classified as an enhancement rather than a bug/feature.

### Summary

# Support `MessageQueueSelector` for transactional messages

## Background

RocketMQ currently supports selecting a target queue for normal messages through APIs such as:

```java
DefaultMQProducer#send(Message msg, MessageQueueSelector selector, Object arg)
```

This is useful for scenarios that require deterministic queue routing, for example ordered messages, sharding by business key, traffic isolation, canary release, or custom queue lane routing.

However, transactional messages currently do not expose an equivalent queue selection API.

The current transactional send API is:

```java
TransactionMQProducer#sendMessageInTransaction(Message msg, Object arg)
```

Internally, `TransactionMQProducer` delegates to `DefaultMQProducerImpl#sendMessageInTransaction(...)`, which marks the message as a prepared transaction message and then calls the default send path:

```java
sendResult = this.send(msg);
```

As a result, transactional messages go through `sendDefaultImpl(...)`, where the client chooses a queue by `selectOneMessageQueue(...)`. There is no way for users to provide a custom `MessageQueueSelector`.

Relevant source references:

- `TransactionMQProducer#sendMessageInTransaction(...)`
-
- `DefaultMQProducerImpl#sendMessageInTransaction(...)`
-
- Normal message selector path:
-
-

## Broker-side behavior

From the broker side, the final queue of a transactional message is determined by the queue selected when the half message is sent.

When the broker receives a prepared transactional message, it stores the original topic and queue id into message properties:

```text
PROPERTY_REAL_TOPIC
PROPERTY_REAL_QUEUE_ID
```

Then the message is rewritten to the internal half topic.

During commit, the broker restores the final message topic and queue id from these properties.

Relevant source references:

- Store original topic and queue id before writing half message:
-
- Restore original topic and queue id when committing:
-

This means queue selection must happen before the half message is sent. After the half message is stored, the commit phase only restores the previously selected queue and does not perform routing again.

## Problem

For transactional messages, users cannot provide a `MessageQueueSelector`, so they cannot control which queue the half message is written to.

This makes it difficult to support scenarios such as:

- deterministic transaction message sharding by business key
- canary / gray release queue lanes
- traffic isolation by queue
- keeping normal messages and transactional messages under the same queue routing strategy

For example, an application may use a custom selector for normal messages to route gray traffic to a subset of queues. But when the same flow uses transactional messages, the routing strategy cannot be reused because transactional send always uses the default queue selection logic.

## Proposal

Add overloaded transactional send APIs that accept `MessageQueueSelector`, similar to normal message sending APIs.

For example:

```java
TransactionSendResult sendMessageInTransaction(
Message msg,
MessageQueueSelector selector,
Object selectorArg,
Object transactionArg
) throws MQClientException;
```

Internally, the transactional send flow could still keep the existing transaction lifecycle unchanged:

1. validate transaction listener
2. mark message with `PROPERTY_TRANSACTION_PREPARED`
3. set producer group
4. send the half message using selector-based queue selection
5. execute local transaction
6. end transaction with commit / rollback / unknown

The only expected difference is step 4: instead of always calling the default send path, allow the half message to be sent through the selector-based path.

## Compatibility

This can be added as an overload, so existing APIs and behavior remain unchanged.

Existing users of:

```java
sendMessageInTransaction(Message msg, Object arg)
```

would continue to use the current default queue selection logic.

Users who need deterministic transactional queue routing can opt in to the new overload.

## Expected behavior

When a transactional message is sent with a `MessageQueueSelector`, the selected queue should become the original queue of the half message. The broker should then store this queue id as `PROPERTY_REAL_QUEUE_ID` and restore it when the transaction is committed, consistent with the existing broker transaction message flow.

## Additional context

This request does not require broker-side transaction protocol changes. The broker already restores the final queue from the queue id selected before the half message is stored. The missing capability is mainly on the client API and transactional send path.

### Motivation

Transactional messages currently do not provide an API to select the target message queue.

For normal messages, RocketMQ supports `MessageQueueSelector`, which allows users to route messages deterministically by business key, shard key, canary/gray label, or other custom routing rules. However, transactional messages always go through the default queue selection path internally.

From the broker-side transaction flow, the final queue of a transactional message is determined when the half message is sent. The broker stores the original queue id as `PROPERTY_REAL_QUEUE_ID` and restores it when the transaction is committed. Therefore, if users cannot select the queue before the half message is written, they cannot control the final queue of the committed transactional message.

This makes it difficult to keep transactional messages and normal messages under the same routing strategy, especially in scenarios such as deterministic sharding, traffic isolation, ordered business lanes, and canary/gray release queue lanes.

### Describe the Solution You'd Like

Transactional messages currently do not provide an API to select the target message queue.

For normal messages, RocketMQ supports `MessageQueueSelector`, which allows users to route messages deterministically by business key, shard key, canary/gray label, or other custom routing rules. However, transactional messages always go through the default queue selection path internally.

From the broker-side transaction flow, the final queue of a transactional message is determined when the half message is sent. The broker stores the original queue id as `PROPERTY_REAL_QUEUE_ID` and restores it when the transaction is committed. Therefore, if users cannot select the queue before the half message is written, they cannot control the final queue of the committed transactional message.

This makes it difficult to keep transactional messages and normal messages under the same routing strategy, especially in scenarios such as deterministic sharding, traffic isolation, ordered business lanes, and canary/gray release queue lanes.

### Describe Alternatives You've Considered

I considered the following alternatives:

1. Put the desired queue id into message properties or user headers.

This does not work because the broker does not read the target queue id from message properties. The queue id is set in `SendMessageRequestHeader` by the client, based on the selected `MessageQueue`.

2. Select or change the queue during transaction commit.

This does not fit the current broker transaction flow. For transactional messages, the broker stores the original queue id as `PROPERTY_REAL_QUEUE_ID` when the half message is written. During commit, the broker only restores the original queue id from the half message. It does not perform queue selection again.

3. Add custom broker-side routing logic based on message properties.

This is possible in theory, but it would push application-level routing rules into the broker. It also introduces compatibility and maintenance risks around half messages, commit/rollback, transaction check, operation messages, and future upgrades.

4. Use normal messages with an outbox pattern instead of RocketMQ transactional messages.

This can be used as an application-level workaround, but it changes the programming model and does not solve the missing queue-selection capability for RocketMQ transactional messages.

### Additional Context

_No response_

Contributor guide

Open the contributing guide

Research direction

Start with TransactionMQProducer#sendMessageInTransaction(...) and DefaultMQProducerImpl#sendMessageInTransaction(...) in the client producer sources, then compare the selector overloads in DefaultMQProducer and its implementation. Add the transactional selector path while preserving the existing overload, and verify that the selected queue is used for the half message and retained through commit.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
api, backend, distributed-systems
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
58/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.