[Bug] CommitLog returns PUT_OK for a message that could not be appended, so a failed send is acknowledged as success

Start in store/src/main/java/org/apache/rocketmq/store/CommitLog.java, in the END_OF_FILE branch of asyncPutMessage and asyncPutMessages, where the retry result is ignored before the unconditional PUT_OK. Check doAppend's size guard and SendMessageProcessor.handlePutMessageResult. Done means regression tests in BatchPutMessageTest and DefaultMessageStoreTest go from PUT_OK to MESSAGE_ILLEGAL. PR #11422 is already open, so reviewing it may be the better entry point.

Before Creating the Bug Report

• I found a bug, not just asking a question, which should be created in GitHub Discussions.
• I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.
• I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Runtime platform environment

OS: any broker whose commit log file is smaller than the message being sent (mappedFileSizeCommitLog below the message length; no configuration validates it against maxMessageSize)

RocketMQ version

branch: develop
Git commit id: a7edeecfb

JDK Version

Compiler: Temurin 1.8.0_504 (target 1.8, same as CI)

Describe the Bug

When CommitLog.asyncPutMessage hits the end of a commit log file it retries the append on a fresh file, but the retry's outcome is only used to decide whether to call onCommitLogAppend — it never reaches the response:

case END_OF_FILE:
    onCommitLogAppend(msg, result, mappedFile);
    unlockMappedFile = mappedFile;
    // Create a new file, re-write the message
    mappedFile = this.mappedFileQueue.getLastMappedFile(0);
    ...
    result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
    if (AppendMessageStatus.PUT_OK.equals(result.getStatus())) {
        onCommitLogAppend(msg, result, mappedFile);
    }
    break;
...
PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);   // unconditional
return handleDiskFlushAndHA(putMessageResult, msg, needAckNums, needHandleHA);

(CommitLog.java:1114-1131 and :1163; the batch path is identical at :1274-1287 and :1320.) The queue offset is not advanced either, because increaseOffset is guarded by AppendMessageStatus.PUT_OK.equals(result.getStatus()) (:1146-1148), so the message really is not in the commit log.

The retry cannot succeed when the message does not fit into an empty file: doAppend returns END_OF_FILE whenever msgLen + END_FILE_MIN_BLANK_LENGTH > mappedFileSizeCommitLog (CommitLog.java:2054-2068), and on a freshly created file the blank space is mappedFileSizeCommitLog. A message larger than the commit log file therefore produces END_OF_FILE twice, the second one is discarded, and the store answers PutMessageStatus.PUT_OK.

That status is what the broker reports to the producer: SendMessageProcessor.handlePutMessageResult turns PUT_OK into ResponseCode.SUCCESS (broker/.../processor/SendMessageProcessor.java:398-401) and sets the msgId from the append result, so the client gets a successful send for a message that was never stored — silent data loss. The sibling branch in the very same switch shows the intended mapping (MESSAGE_SIZE_EXCEEDED/PROPERTIES_SIZE_EXCEEDED → MESSAGE_ILLEGAL, UNKNOWN_ERROR → UNKNOWN_ERROR, :1132-1137); the retry path just has no equivalent.

The configuration is reachable and used in-tree: this repository's own store tests run with mappedFileSizeCommitLog = 1024 * 8 (store/src/test/java/org/apache/rocketmq/store/BatchPutMessageTest.java:72, and DefaultMessageStoreTest.java:122), and nothing validates the relation between mappedFileSizeCommitLog and maxMessageSize (default 4 MB) — so a 16 KB message against an 8 KB commit log is accepted by the encoder and then lost with a success ack.

Steps to Reproduce

1. Start a broker (or build a DefaultMessageStore in a test) with mappedFileSizeCommitLog smaller than the message body, e.g. mappedFileSizeCommitLog=8192, and keep the default maxMessageSize.
2. Send a message whose encoded length exceeds the file size but is below maxMessageSize (e.g. a 16 KB body).
3. store.putMessage(...) returns PutMessageResult with putMessageStatus = PUT_OK although appendMessageResult.getStatus() is END_OF_FILE; the commit log stays empty and no consume-queue entry is written.

What Did You Expect to See?

The broker reports the append failure to the producer (the same way the first-append branch reports MESSAGE_ILLEGAL for a message the store cannot accept), so the client sees an error instead of a success acknowledgement for a message that was dropped.

What Did You See Instead?

PutMessageStatus.PUT_OK for a message that was never written: SEND_OK on the producer side, a msgId computed for the never-written file, and inflated put-message statistics, while the message cannot be consumed because it does not exist.

Additional Context

Minimal fix: after the retry append in both asyncPutMessage and asyncPutMessages, map a non-PUT_OK retry result the same way the first-append branch does — UNKNOWN_ERROR stays UNKNOWN_ERROR, everything else (END_OF_FILE, MESSAGE_SIZE_EXCEEDED, PROPERTIES_SIZE_EXCEEDED) becomes MESSAGE_ILLEGAL — and return it instead of falling through to the unconditional PUT_OK.

I will open a PR with the change plus regression tests for both the single-message and the batch path (red before: PUT_OK/isOk() == true for a message larger than the commit log file; green after: MESSAGE_ILLEGAL).