Skip to content

RATIS-2282. LogAppender Restart Due to Premature Log Entry Access During Concurrent Write Processing #1249

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

Merged
szetszwo merged 2 commits into from
Apr 20, 2025
Merged

RATIS-2282. LogAppender Restart Due to Premature Log Entry Access During Concurrent Write Processing #1249

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4045ca9
Put entry to cache before adding the record
RkGrit Apr 19, 2025
4500bd9
modify getLogRecord()
RkGrit Apr 20, 2025
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
32 changes: 15 additions & 17 deletions ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/LogSegment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -454,10 +454,13 @@ private void append(Op op, ReferenceCountedObject<LogEntryProto> entryRef,
boolean keepEntryInCache, Consumer<LogEntryProto> logConsumer) {
final LogEntryProto entry = entryRef.retain();
try {
final LogRecord record = appendLogRecord(op, entry);
final LogRecord record = new LogRecord(totalFileSize, entry);
if (keepEntryInCache) {
putEntryCache(record.getTermIndex(), entryRef, op);
}
appendLogRecord(op, record);
totalFileSize += getEntrySize(entry, op);

if (logConsumer != null) {
logConsumer.accept(entry);
}
Expand All @@ -466,24 +469,22 @@ private void append(Op op, ReferenceCountedObject<LogEntryProto> entryRef,
}
}


private LogRecord appendLogRecord(Op op, LogEntryProto entry) {
Objects.requireNonNull(entry, "entry == null");
private void appendLogRecord(Op op, LogRecord record) {
Objects.requireNonNull(record, "record == null");
final LogRecord currentLast = records.getLast();

final long index = record.getTermIndex().getIndex();
if (currentLast == null) {
Preconditions.assertTrue(entry.getIndex() == startIndex,
"gap between start index %s and first entry to append %s",
startIndex, entry.getIndex());
Preconditions.assertTrue(index == startIndex,
"%s: gap between start index %s and the entry to append %s", op, startIndex, index);
} else {
Preconditions.assertTrue(entry.getIndex() == currentLast.getTermIndex().getIndex() + 1,
"gap between entries %s and %s", entry.getIndex(), currentLast.getTermIndex().getIndex());
final long currentLastIndex = currentLast.getTermIndex().getIndex();
Preconditions.assertTrue(index == currentLastIndex + 1,
"%s: gap between last entry %s and the entry to append %s", op, currentLastIndex, index);
}

final LogRecord record = new LogRecord(totalFileSize, entry);
records.append(record);
totalFileSize += getEntrySize(entry, op);
endIndex = entry.getIndex();
return record;
endIndex = index;
}

ReferenceCountedObject<LogEntryProto> getEntryFromCache(TermIndex ti) {
Expand Down Expand Up @@ -514,10 +515,7 @@ synchronized ReferenceCountedObject<LogEntryProto> loadCache(TermIndex ti) throw
}

LogRecord getLogRecord(long index) {
if (index >= startIndex && index <= endIndex) {
return records.get(index);
}
return null;
return records.get(index);
}

TermIndex getLastTermIndex() {
Expand Down
Loading