Skip to content

[ISSUE #10755] Fix ConsumeQueueExt truncation cleanup #10758

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
ai-yang wants to merge 1 commit into
base: develop
Choose a base branch
from
+607 −38
Open

[ISSUE #10755] Fix ConsumeQueueExt truncation cleanup #10758

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
117 changes: 79 additions & 38 deletions store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,15 @@ public boolean load() {
@Override
public void recover() {
final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
if (!mappedFiles.isEmpty()) {
if (mappedFiles.isEmpty()) {
if (isExtReadEnable()) {
this.consumeQueueExt.recover();
if (!this.consumeQueueExt.truncateAll()) {
log.warn("Failed to truncate all consume queue ext data during recovery, topic={}, queueId={}",
this.topic, this.queueId);
}
}
} else {

int index = mappedFiles.size() - 3;
if (index < 0) {
Expand Down Expand Up @@ -423,10 +431,12 @@ public void truncateDirtyLogicFiles(long phyOffset, boolean deleteFile) {

this.setMaxPhysicOffset(phyOffset);
long maxExtAddr = 1;
boolean shouldDeleteFile = false;
boolean hasRetainedExt = false;
boolean cqFileDeletionFailed = false;
while (true) {
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
if (mappedFile != null) {
boolean shouldDeleteFile = false;
ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();

mappedFile.setWrotePosition(0);
Expand All @@ -438,63 +448,94 @@ public void truncateDirtyLogicFiles(long phyOffset, boolean deleteFile) {
int size = byteBuffer.getInt();
long tagsCode = byteBuffer.getLong();

if (0 == i) {
if (offset >= phyOffset) {
if (offset < 0 || size <= 0 || offset >= phyOffset) {
if (0 == i) {
shouldDeleteFile = true;
break;
} else {
int pos = i + CQ_STORE_UNIT_SIZE;
mappedFile.setWrotePosition(pos);
mappedFile.setCommittedPosition(pos);
mappedFile.setFlushedPosition(pos);
this.setMaxPhysicOffset(offset + size);
// This maybe not take effect, when not every consume queue has extend file.
if (isExtAddr(tagsCode)) {
maxExtAddr = tagsCode;
}
}
} else {

if (offset >= 0 && size > 0) {

if (offset >= phyOffset) {
return;
}
break;
}

int pos = i + CQ_STORE_UNIT_SIZE;
mappedFile.setWrotePosition(pos);
mappedFile.setCommittedPosition(pos);
mappedFile.setFlushedPosition(pos);
this.setMaxPhysicOffset(offset + size);
if (isExtAddr(tagsCode)) {
maxExtAddr = tagsCode;
}
int pos = i + CQ_STORE_UNIT_SIZE;
mappedFile.setWrotePosition(pos);
mappedFile.setCommittedPosition(pos);
mappedFile.setFlushedPosition(pos);
this.setMaxPhysicOffset(offset + size);
// This maybe not take effect, when not every consume queue has extend file.
long logicOffset = mappedFile.getFileFromOffset() + i;
if (logicOffset >= this.minLogicOffset && isExtAddr(tagsCode)) {
maxExtAddr = tagsCode;
hasRetainedExt = true;
}

if (pos == logicFileSize) {
return;
}
} else {
return;
}
if (pos == logicFileSize) {
break;
}
}

if (shouldDeleteFile) {
if (deleteFile) {
String mappedFilePath = mappedFile.getFileName();
this.mappedFileQueue.deleteLastMappedFile();
if (new File(mappedFilePath).exists()) {
cqFileDeletionFailed = true;
log.warn("Consume queue file still exists after deletion: {}", mappedFilePath);
}
} else {
this.mappedFileQueue.deleteExpiredFile(Collections.singletonList(this.mappedFileQueue.getLastMappedFile()));
}
} else {
break;
}

} else {
break;
}
}

if (isExtReadEnable()) {
this.consumeQueueExt.truncateByMaxAddress(maxExtAddr);
if (deleteFile && isExtReadEnable()) {
if (cqFileDeletionFailed) {
log.warn("Skip truncating consume queue ext because a consume queue file was not deleted");
return;
}
if (hasRetainedExt && this.consumeQueueExt.get(maxExtAddr) == null) {
hasRetainedExt = false;
}
if (!hasRetainedExt) {
maxExtAddr = findLastRetainedExtAddress();
hasRetainedExt = isExtAddr(maxExtAddr);
}
if (hasRetainedExt) {
this.consumeQueueExt.truncateByMaxAddress(maxExtAddr);
} else {
if (!this.consumeQueueExt.truncateAll()) {
log.warn("Failed to truncate all consume queue ext data, topic={}, queueId={}",
this.topic, this.queueId);
}
}
}
}

private long findLastRetainedExtAddress() {
List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
for (int fileIndex = mappedFiles.size() - 1; fileIndex >= 0; fileIndex--) {
MappedFile mappedFile = mappedFiles.get(fileIndex);
ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
for (int position = mappedFile.getWrotePosition() - CQ_STORE_UNIT_SIZE;
position >= 0; position -= CQ_STORE_UNIT_SIZE) {
long logicOffset = mappedFile.getFileFromOffset() + position;
if (logicOffset < this.minLogicOffset) {
return 1;
}
long offset = byteBuffer.getLong(position);
int size = byteBuffer.getInt(position + Long.BYTES);
long tagsCode = byteBuffer.getLong(position + MSG_TAG_OFFSET_INDEX);
if (offset >= 0 && size > 0 && isExtAddr(tagsCode)
&& this.consumeQueueExt.get(tagsCode) != null) {
return tagsCode;
}
}
}
return 1;
}

@Override
Expand Down
34 changes: 34 additions & 0 deletions store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ public class ConsumeQueueExt {
private final String storePath;
private final int mappedFileSize;
private ByteBuffer tempContainer;
private volatile boolean truncateAllPending;

public static final int END_BLANK_DATA_LENGTH = 4;

Expand Down Expand Up @@ -228,6 +229,11 @@ public boolean get(final long address, final CqExtUnit cqExtUnit) {
* @return success: < 0: fail: >=0
*/
public long put(final CqExtUnit cqExtUnit) {
if (this.truncateAllPending) {
log.warn("Skip saving consume queue ext while truncating all data is pending, {}", cqExtUnit);
return 1;
}

final int retryTimes = 3;
try {
int size = cqExtUnit.calcUnitSize();
Expand Down Expand Up @@ -405,6 +411,34 @@ public void truncateByMaxAddress(final long maxAddress) {
this.mappedFileQueue.truncateDirtyFiles(realOffset + cqExtUnit.getSize());
}

/**
* Delete all consume queue extension data when no consume queue entry retains an extension address.
*/
public synchronized boolean truncateAll() {
log.info("Truncate all consume queue ext data.");
this.truncateAllPending = true;
List<MappedFile> deletedFiles = new ArrayList<>();
for (MappedFile mappedFile : new ArrayList<>(this.mappedFileQueue.getMappedFiles())) {
boolean destroyed = mappedFile.destroy(1000 * 3);
boolean fileExists = new File(mappedFile.getFileName()).exists();
if (destroyed && !fileExists) {
deletedFiles.add(mappedFile);
} else {
log.warn("Consume queue ext file remains after truncating all data, file={}, destroyed={}",
mappedFile.getFileName(), destroyed);
}
}
this.mappedFileQueue.deleteExpiredFile(deletedFiles);
if (!this.mappedFileQueue.getMappedFiles().isEmpty()) {
return false;
}

this.mappedFileQueue.setFlushedWhere(0);
this.mappedFileQueue.setCommittedWhere(0);
this.truncateAllPending = false;
return true;
}

/**
* flush buffer to file.
*/
Expand Down
Loading