Transactional Messaging
Core Concepts
Transactional messaging in RocketMQ implements the local transaction table pattern from distributed systems, substituting message middleware for database operations while handling status verification automatically.
Unlike Apache Kafka's transaction implementation focused on Exactly-Once semantics for stream processing, RocketMQ's approach centers on ensuring transaction completion between producers and brokers.
Processing Workflow
- Producer sends a preparation message to broker
- Broker persists the preparation message
- Based on local transaction outcome, producer sends either commit or rollback instruction
- On commit receipt, broker makes message visible to consumers; on rollback, discards the message
Recovery Mechanism
When broker doesn't receive transaction results within timeout period:
- Initiates transaction status inquiry to producer
- Producer executes callback method to determine actual transaction state
- Broker processes according to returned commit/rollback decision
Implementation Details
After configuring transaction listeners (execution handler and status checker), producers initiate transactional message delivery. The system marks messages with transaction flags before standard synchronous transmission.
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
sendResult = this.send(msg);
Upon broker reception, messages follow regular processing paths but undergo special handling via TransactionalMessageService. The service stores original topic/queue information and redirects messages to internal transaction topic (RMQ_SYS_TRANS_HALF_TOPIC).
private MessageExtBrokerInner processPreparationMessage(MessageExtBrokerInner incomingMsg) {
MessageAccessor.putProperty(incomingMsg, MessageConst.PROPERTY_REAL_TOPIC, incomingMsg.getTopic());
MessageAccessor.putProperty(incomingMsg, MessageConst.PROPERTY_REAL_QUEUE_ID,
String.valueOf(incomingMsg.getQueueId()));
incomingMsg.setSysFlag(
MessageSysFlag.resetTransactionValue(incomingMsg.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
incomingMsg.setTopic(TransactionalMessageUtil.buildHalfTopic());
incomingMsg.setQueueId(0);
incomingMsg.setPropertiesString(MessageDecoder.messageProperties2String(incomingMsg.getProperties()));
return incomingMsg;
}
Post-transmission, producer executes local business logic using dedicated thread pool:
localTxState = localTxExecutor.executeLocalTransactionBranch(msg, arg);
this.reportTransactionStatus(msg, sendResult, localTxState, localException);
Reporting uses one-way communication with specific request code:
public static final int REPORT_TRANSACTION_RESULT = 37;
Broker-side handling occurs through EndTransactionProcessor which processes commit/rollback decisions differently:
if (MessageSysFlag.TRANSACTION_COMMIT_TYPE == requestHeader.getTransactionDecision()) {
ProcessResult result = this.brokerController.getTransactionalMessageService().commit(requestHeader);
if (result.getResponseCode() == ResponseCode.SUCCESS) {
if (res.getCode() == ResponseCode.SUCCESS) {
RemotingCommand deliveryOutcome = dispatchActualMessage(restoredMessage);
if (deliveryOutcome.getCode() == ResponseCode.SUCCESS) {
this.brokerController.getTransactionalMessageService()
.markPreparationComplete(result.getPrepareMessage());
}
return deliveryOutcome;
}
return res;
}
}
Rollback operations mark preparations for deletion by writing transaction IDs to special operation topic:
if (MessageSysFlag.TRANSACTION_ROLLBACK_TYPE == requestHeader.getTransactionDecision()) {
ProcessResult result = this.brokerController.getTransactionalMessageService().rollback(requestHeader);
if (result.getResponseCode() == ResponseCode.SUCCESS) {
RemotingCommand validation = validatePreparation(result.getPrepareMessage(), requestHeader);
if (validation.getCode() == ResponseCode.SUCCESS) {
this.brokerController.getTransactionalMessageService()
.markPreparationComplete(result.getPrepareMessage());
}
return validation;
}
}
Status Verification Process
Broker initiates periodic checks when transactions remain unresolved beyond configured timeouts. During startup, BrokerController launches transaction monitoring service executing continuous scans via TransactionalMessageServiceImpl.verify() method.
Core verification logic examines preparation queues:
String transactionTopic = TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC;
Set<MessageQueue> preparationQueues = txMessageBridge.fetchMessageQueues(transactionTopic);
for (MessageQueue queue : preparationQueues) {
long scanStartTime = System.currentTimeMillis();
MessageQueue operationQueue = getOperationQueue(queue);
long preparationOffset = txMessageBridge.fetchConsumeOffset(queue);
long operationOffset = txMessageBridge.fetchConsumeOffset(operationQueue);
List<Long> processedOperations = new ArrayList<>();
HashMap<Long, Long> completedMappings = new HashMap<>();
PullResult pullOutcome = populateOperationMap(completedMappings, operationQueue,
operationOffset, preparationOffset, processedOperations);
Operation mapping identifies resolved transactions:
private PullResult populateOperationMap(HashMap<Long, Long> resolutionMap,
MessageQueue opQueue, long opStartOffset,
long minPrepOffset, List<Long> processedOps) {
PullResult outcome = pullOperations(opQueue, opStartOffset, 32);
List<MessageExt> operationMessages = outcome.getMsgFoundList();
for (MessageExt operationMsg : operationMessages) {
Long prepOffset = Long.valueOf(new String(operationMsg.getBody(), TransactionalMessageUtil.charset));
if (TransactionalMessageUtil.RESOLUTION_TAG.equals(operationMsg.getTags())) {
if (prepOffset < minPrepOffset) {
processedOps.add(operationMsg.getQueueOffset());
} else {
resolutionMap.put(prepOffset, operationMsg.getQueueOffset());
}
}
}
return outcome;
}
Verification continues examining unresolved preparations:
while (true) {
if (System.currentTimeMillis() - scanStartTime > MAX_PROCESSING_DURATION) break;
if (resolutionMap.containsKey(currentIndex)) {
Long resolvedOpOffset = resolutionMap.remove(currentIndex);
processedOperations.add(resolvedOpOffset);
} else {
GetResult prepResult = getPreparation(queue, currentIndex);
if (shouldDiscard(prepMsg, maxRetries) || shouldSkip(prepMsg)) {
listener.handleExpiredMessage(prepMsg);
newOffset = currentIndex + 1;
currentIndex++;
continue;
}
Immunity periods prevent premature verification attempts:
long elapsedSinceCreation = System.currentTimeMillis() - prepMsg.getBornTimestamp();
long immunityThreshold = defaultTimeout;
String customImmunity = prepMsg.getUserProperty(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS);
if (customImmunity != null) {
immunityThreshold = calculateImmunity(customImmunity, defaultTimeout);
if (elapsedSinceCreation < immunityThreshold) {
if (verifyQueuePosition(resolutionMap, processedOperations, prepMsg)) {
newOffset = currentIndex + 1;
currentIndex++;
continue;
}
}
} else {
if (elapsedSinceCreation >= 0 && elapsedSinceCreation < immunityThreshold) {
break;
}
}
Verification triggers under three conditions:
- Empty operation set exceeding immunity period
- Latest operation timestamp surpassing timeout threshold
- Disabled immunity protection
Producer-side handling uses ClientRemotingProcessor.verifyTransactionState():
String txId = messageExt.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
if (txId != null && !txId.isEmpty()) {
messageExt.setTransactionId(txId);
}
final String producerGroup = messageExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);
if (producerGroup != null) {
MQProducerInner producer = this.mqClientFactory.selectProducer(producerGroup);
if (producer != null) {
final String sourceAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
producer.verifyTransactionState(sourceAddress, messageExt, requestHeader);
}
Producer verification executes registered callbacks:
TransactionCheckListener txChecker = DefaultMQProducerImpl.this.statusVerifier;
TransactionListener txListener = getStatusChecker();
if (txChecker != null || txListener != null) {
LocalTransactionState txState = LocalTransactionState.UNKNOWN;
Throwable error = null;
try {
if (txChecker != null) {
txState = txChecker.verifyLocalTransactionState(message);
} else if (txListener != null) {
txState = txListener.checkLocalTransaction(message);
}
} catch (Throwable e) {
error = e;
}
this.processTransactionVerification(txState, group, error);
}
Batch Messaging
Design Principles
Batch messaging optimizes bandwidth usage and reduces header overhead by combining multiple messages into single transmissions.
Processing Pipeline
Batch submission requires validation ensuring consistent topics and no retry destinations:
List<Message> validatedMessages = new ArrayList<>(inputMessages.size());
Message referenceMessage = null;
for (Message candidate : inputMessages) {
if (candidate.getDelayTimeLevel() > 0) {
throw new UnsupportedOperationException("Delayed batches unsupported");
}
if (candidate.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
throw new UnsupportedOperationException("Retry topic batches unsupported");
}
if (referenceMessage == null) {
referenceMessage = candidate;
} else {
if (!referenceMessage.getTopic().equals(candidate.getTopic())) {
throw new UnsupportedOperationException("Mixed topics in batch");
}
if (referenceMessage.isWaitStoreMsgOK() != candidate.isWaitStoreMsgOK()) {
throw new UnsupportedOperationException("Inconsistent storage confirmation settings");
}
}
validatedMessages.add(candidate);
}
MessageBatch batchContainer = new MessageBatch(validatedMessages);
Encoding combines individual message representations:
public static byte[] encodeBatch(List<Message> messages) {
List<byte[]> encodedIndividuals = new ArrayList<>(messages.size());
int totalSize = 0;
for (Message item : messages) {
byte[] encodedItem = encodeSingle(item);
encodedIndividuals.add(encodedItem);
totalSize += encodedItem.length;
}
byte[] combinedBuffer = new byte[totalSize];
int writePosition = 0;
for (byte[] fragment : encodedIndividuals) {
System.arraycopy(fragment, 0, combinedBuffer, writePosition, fragment.length);
writePosition += fragment.length;
}
return combinedBuffer;
}
Broker processing splits batches during persistence using dedicated request code:
public static final int BATCH_MESSAGE_SUBMISSION = 88;
Storage decomposes batches into individual entries:
while (batchBuffer.hasRemaining()) {
final int startPosition = batchBuffer.position();
final int messageLength = batchBuffer.getInt();
// Standard message storage processing
queueOffset++;
messageCount++;
batchBuffer.position(startPosition + messageLength);
}
Delayed Messaging
Operational Model
Delayed messaging enables scheduled task execution where messages become consumer-visible after specified intervals. RocketMQ supports predefined delay levels rather than arbitrary timing.
Implementation Architecture
Broker-side processing intercepts delayed messages during commit log insertion:
if (msg.getDelayTimeLevel() > 0) {
if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
}
targetTopic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
targetQueue = ScheduleMessageService.mapDelayLevelToQueue(msg.getDelayTimeLevel());
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
msg.setTopic(targetTopic);
msg.setQueueId(targetQueue);
}
ScheduleMessageService manages timed delivery through initialization:
public void initialize() {
if (started.compareAndSet(false, true)) {
super.load();
this.scheduler = new Timer("ScheduledMessageTimer", true);
for (Map.Entry<Integer, Long> delayEntry : this.levelMapping.entrySet()) {
Integer level = delayEntry.getKey();
Long delayDuration = delayEntry.getValue();
Long consumptionOffset = this.offsetTracking.get(level);
if (consumptionOffset == null) {
consumptionOffset = 0L;
}
if (delayDuration != null) {
this.scheduler.schedule(new DeliveryTimerTask(level, consumptionOffset), INITIAL_DELAY);
}
}
this.scheduler.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
if (started.get()) persistOffsets();
} catch (Throwable e) {
log.error("Offset persistence failure", e);
}
}
}, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval());
}
}
Per-level scheduling executes delivery checks:
public void executeDeliveryCheck() {
ConsumeQueue scheduledQueue = findScheduledQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
mapDelayLevelToQueue(delayLevel));
long previousOffset = lastProcessedOffset;
if (scheduledQueue != null) {
SelectMappedBufferResult indexBuffer = scheduledQueue.getIndexBuffer(this.lastProcessedOffset);
if (indexBuffer != null) {
try {
long nextPosition = lastProcessedOffset;
int positionIndex = 0;
ConsumeQueueExt.CqExtUnit extensionUnit = new ConsumeQueueExt.CqExtUnit();
for (; positionIndex < indexBuffer.getSize(); positionIndex += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
long physicalOffset = indexBuffer.getByteBuffer().getLong();
int physicalSize = indexBuffer.getByteBuffer().getInt();
long tagIdentifier = indexBuffer.getByteBuffer().getLong();
long currentTime = System.currentTimeMillis();
long scheduledTime = adjustScheduledTimestamp(currentTime, tagIdentifier);
nextPosition = lastProcessedOffset + (positionIndex / ConsumeQueue.CQ_STORE_UNIT_SIZE);
long remainingWait = scheduledTime - currentTime;
if (remainingWait <= 0) {
MessageExt scheduledMsg = defaultMessageStore.lookMessageByOffset(physicalOffset, physicalSize);
if (scheduledMsg != null) {
try {
MessageExtBrokerInner restoredMsg = restoreOriginalMessage(scheduledMsg);
PutMessageResult storeResult = defaultMessageStore.putMessage(restoredMsg);
} catch (Exception e) {
// Error handling
}
}
} else {
scheduler.schedule(new DeliveryTimerTask(this.delayLevel, nextPosition), remainingWait);
updateOffset(this.delayLevel, nextPosition);
return;
}
}
nextPosition = lastProcessedOffset + (positionIndex / ConsumeQueue.CQ_STORE_UNIT_SIZE);
scheduler.schedule(new DeliveryTimerTask(this.delayLevel, nextPosition), DEFAULT_WAIT_PERIOD);
updateOffset(this.delayLevel, nextPosition);
return;
} finally {
indexBuffer.release();
}
}
}
scheduler.schedule(new DeliveryTimerTask(this.delayLevel, previousOffset), DEFAULT_WAIT_PERIOD);
}