Skip to content

[ISSUE #10666][BUG] Persist commitLog truncate marker #10816

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chenxu80 wants to merge 1 commit into
base: develop
Choose a base branch
from
+169 −5
Open

[ISSUE #10666][BUG] Persist commitLog truncate marker #10816

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 70 additions & 0 deletions store/src/main/java/org/apache/rocketmq/store/CommitLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.net.Inet6Address;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
Expand Down Expand Up @@ -899,6 +900,75 @@ public void truncateDirtyFiles(long phyOffset) {
}
}

void persistTruncateMarker(long phyOffset) {
MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(phyOffset, false);
if (mappedFile == null) {
throw new IllegalStateException("Cannot find mapped file when persisting truncate marker, offset="
+ phyOffset);
}

int relativeOffset = (int) (phyOffset - mappedFile.getFileFromOffset());
if (relativeOffset < 0 || relativeOffset + 2 * Integer.BYTES > mappedFile.getFileSize()) {
throw new IllegalStateException("Invalid truncate marker position, offset=" + phyOffset
+ ", file=" + mappedFile.getFileName() + ", relativeOffset=" + relativeOffset);
}

if (!mappedFile.hold()) {
throw new IllegalStateException("Cannot hold mapped file when persisting truncate marker, offset="
+ phyOffset + ", file=" + mappedFile.getFileName());
}

try {
MappedByteBuffer mappedByteBuffer = mappedFile.getMappedByteBuffer();
validateTruncateMarkerPosition(mappedByteBuffer, mappedFile, phyOffset, relativeOffset);
// An invalid zero header makes recovery stop here instead of rolling to the next mapped file.
// Do not use appendMessage because persisting the marker must not change the logical wrote position.
mappedByteBuffer.putLong(relativeOffset, 0L);
mappedByteBuffer.force();
log.info("Persisted CommitLog truncate marker, offset={}, file={}, relativeOffset={}",
phyOffset, mappedFile.getFileName(), relativeOffset);
} catch (RuntimeException e) {
throw new IllegalStateException("Failed to persist CommitLog truncate marker, offset=" + phyOffset
+ ", file=" + mappedFile.getFileName(), e);
} finally {
mappedFile.release();
}
}

private void validateTruncateMarkerPosition(ByteBuffer buffer, MappedFile mappedFile, long phyOffset,
int relativeOffset) {
int currentPosition = 0;
while (currentPosition < relativeOffset) {
if (currentPosition + 2 * Integer.BYTES > relativeOffset) {
throw invalidTruncateMarkerPosition(mappedFile, phyOffset, relativeOffset, currentPosition);
}

int totalSize = buffer.getInt(currentPosition);
int magicCode = buffer.getInt(currentPosition + Integer.BYTES);
boolean invalidMagicCode = magicCode != MessageDecoder.MESSAGE_MAGIC_CODE
&& magicCode != MessageDecoder.MESSAGE_MAGIC_CODE_V2;
if (invalidMagicCode
|| totalSize <= MessageDecoder.MESSAGE_PHYSIC_OFFSET_POSITION + Long.BYTES
|| (long) currentPosition + totalSize > relativeOffset
|| buffer.getLong(currentPosition + MessageDecoder.MESSAGE_PHYSIC_OFFSET_POSITION)
!= mappedFile.getFileFromOffset() + currentPosition) {
throw invalidTruncateMarkerPosition(mappedFile, phyOffset, relativeOffset, currentPosition);
}
currentPosition += totalSize;
}

if (currentPosition != relativeOffset) {
throw invalidTruncateMarkerPosition(mappedFile, phyOffset, relativeOffset, currentPosition);
}
}

private IllegalStateException invalidTruncateMarkerPosition(MappedFile mappedFile, long phyOffset,
int relativeOffset, int currentPosition) {
return new IllegalStateException("Truncate marker offset is not a CommitLog message boundary, offset="
+ phyOffset + ", file=" + mappedFile.getFileName() + ", relativeOffset=" + relativeOffset
+ ", validationPosition=" + currentPosition);
}

protected void onCommitLogAppend(MessageExtBrokerInner msg, AppendMessageResult result, MappedFile commitLogFile) {
this.getMessageStore().onCommitLogAppend(msg, result, commitLogFile);
}
Expand Down
19 changes: 14 additions & 5 deletions store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -793,17 +793,26 @@ public void truncateDirtyFiles(long offsetToTruncate) throws RocksDBException {
return;
}

long markerOffset = Math.max(offsetToTruncate, this.getMinPhyOffset());
this.reputMessageService.shutdown();

long oldReputFromOffset = this.reputMessageService.getReputFromOffset();

// truncate consume queue
this.truncateDirtyLogicFiles(offsetToTruncate);
try {
this.commitLog.persistTruncateMarker(markerOffset);

// truncate commitLog
this.commitLog.truncateDirtyFiles(offsetToTruncate);
// truncate consume queue
this.truncateDirtyLogicFiles(offsetToTruncate);

this.recoverTopicQueueTable();
// truncate commitLog
this.commitLog.truncateDirtyFiles(offsetToTruncate);

this.recoverTopicQueueTable();
} catch (RocksDBException | RuntimeException e) {
LOGGER.error("Failed to truncate dirty files to {} after stopping ReputMessageService; "
+ "keep the service stopped to avoid processing a partially truncated CommitLog", offsetToTruncate, e);
throw e;
}

if (!messageStoreConfig.isEnableBuildConsumeQueueConcurrently()) {
this.reputMessageService = new ReputMessageService();
Expand Down
85 changes: 85 additions & 0 deletions store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.apache.rocketmq.store.exception.ConsumeQueueException;
import org.apache.rocketmq.store.logfile.MappedFile;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
Expand Down Expand Up @@ -728,6 +729,90 @@ public void testRecover() throws Exception {
}
}

@Test
public void testTruncateMarkerSurvivesNormalRecovery() throws Exception {
verifyTruncateMarkerRecovery(false, false);
}

@Test
public void testTruncateMarkerSurvivesAbnormalRecovery() throws Exception {
verifyTruncateMarkerRecovery(true, false);
}

@Test
public void testTruncateMarkerAtFirstMessageSurvivesAbnormalRecovery() throws Exception {
verifyTruncateMarkerRecovery(true, true);
}

@Test
public void testPersistTruncateMarkerRejectsNonMessageBoundary() {
queueTotal = 1;
messageBody = storeMessage.getBytes(StandardCharsets.UTF_8);
PutMessageResult firstResult = messageStore.putMessage(buildMessage());
PutMessageResult secondResult = messageStore.putMessage(buildMessage());
assertThat(firstResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);
assertThat(secondResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);

long secondMessageOffset = secondResult.getAppendMessageResult().getWroteOffset();
IllegalStateException exception = Assert.assertThrows(IllegalStateException.class,
() -> getDefaultMessageStore().getCommitLog().persistTruncateMarker(secondMessageOffset + 1));
assertThat(exception.getMessage()).contains("Failed to persist CommitLog truncate marker");
assertThat(exception.getCause().getMessage()).contains("not a CommitLog message boundary");
}

private void verifyTruncateMarkerRecovery(boolean recoverAbnormally, boolean truncateFirstMessage)
throws Exception {
queueTotal = 1;
messageBody = storeMessage.getBytes(StandardCharsets.UTF_8);
PutMessageResult firstResult = messageStore.putMessage(buildMessage());
PutMessageResult secondResult = messageStore.putMessage(buildMessage());
PutMessageResult thirdResult = messageStore.putMessage(buildMessage());
assertThat(firstResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);
assertThat(secondResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);
assertThat(thirdResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);

long firstMessageOffset = firstResult.getAppendMessageResult().getWroteOffset();
long truncateOffset = truncateFirstMessage ? firstMessageOffset
: secondResult.getAppendMessageResult().getWroteOffset();
assertThat(messageStore.truncateFiles(truncateOffset)).isTrue();
assertThat(messageStore.getMaxPhyOffset()).isEqualTo(truncateOffset);

DefaultMessageStore defaultMessageStore = getDefaultMessageStore();
MappedFile mappedFile = defaultMessageStore.getCommitLog().getMappedFileQueue()
.findMappedFileByOffset(truncateOffset, false);
assertThat(mappedFile).isNotNull();
int relativeOffset = (int) (truncateOffset - mappedFile.getFileFromOffset());
assertThat(mappedFile.getMappedByteBuffer().getLong(relativeOffset)).isZero();

String storeRootDir = defaultMessageStore.getMessageStoreConfig().getStorePathRootDir();
messageStore.shutdown();
if (recoverAbnormally) {
File abortFile = new File(StorePathConfigHelper.getAbortFile(storeRootDir));
UtilAll.ensureDirOK(abortFile.getParent());
assertThat(abortFile.createNewFile()).isTrue();
}

MessageStore recoveredMessageStore = null;
try {
recoveredMessageStore = buildMessageStore(storeRootDir);
assertThat(recoveredMessageStore.load()).isTrue();
recoveredMessageStore.start();
assertThat(recoveredMessageStore.getMaxPhyOffset()).isEqualTo(truncateOffset);
if (!truncateFirstMessage) {
assertThat(recoveredMessageStore.lookMessageByOffset(firstMessageOffset)).isNotNull();
}
assertThat(recoveredMessageStore.lookMessageByOffset(truncateOffset)).isNull();
} finally {
if (recoveredMessageStore != null) {
try {
recoveredMessageStore.shutdown();
} finally {
recoveredMessageStore.destroy();
}
}
}
}

@Test
public void testStorePathOK() {
if (messageStore instanceof DefaultMessageStore) {
Expand Down
Loading