8000 fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] by giuseppelillo · Pull Request #857 · aiven/inkless · GitHub
[go: up one dir, main page]
Skip to content

fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] #857

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
giuseppelillo wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] #857

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d2c105f
fix(inkless:consolidation): bound the diskless start by the seal befo…
giuseppelillo Nov 5, 2026
50e0d65
fix(inkless:switch): recreate phase 2 replicas that don't hold the co…
giuseppelillo Nov 6, 2026
037afb6
fix(inkless:consolidation): keep referenced remote segments while the…
giuseppelillo Nov 6, 2026
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
5 changes: 3 additions & 2 deletions INKLESS_OWNERSHIP
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,9 +65,10 @@ core/src/main/scala/kafka/server/DynamicBrokerConfig.scala @aiven/inkless @apach
core/src/main/scala/kafka/server/ConfigHandler.scala @aiven/inkless @apache/kafka # topic-config propagation
core/src/main/scala/kafka/server/ControllerServer.scala @aiven/inkless @apache/kafka # controller-side diskless wiring
core/src/main/scala/kafka/server/AbstractFetcherThread.scala @aiven/inkless @apache/kafka # consolidation fetcher skips ConsumerLag JMX registration
core/src/main/java/kafka/server/TierStateMachine.java @aiven/inkless @apache/kafka # phase 2 remote rebuild starts at the classic-to-diskless seal
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @aiven/inkless @apache/kafka # diskless placement, election, and classic-to-diskless switch commits
metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java @aiven/inkless @apache/kafka # classic-to-diskless switch flag commit
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java @aiven/inkless @apache/kafka # forced deletion of the local classic prefix
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java @aiven/inkless @apache/kafka # forced deletion of the local classic prefix; truncation count that fences remote offset updates
core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala @aiven/inkless @apache/kafka # future log rolls at the seal; promotion waits for classic prefix deletion
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java @aiven/inkless @apache/kafka # remote read throws LeaderEpochNotAvailableException when the epoch cache has no epoch
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java @aiven/inkless @apache/kafka # remote read throws LeaderEpochNotAvailableException when the epoch cache has no epoch; remote offset updates skip a log that truncated during the task; epoch cleanup keeps consolidating segments from the cross-tier log start
core/src/main/scala/kafka/cluster/Partition.scala @aiven/inkless @apache/kafka # classic-to-diskless seal; ISR checks capped at the seal in phase 1
5 changes: 4 additions & 1 deletion clients/src/main/resources/common/message/SetClassicPrefixInRemoteStorageRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,10 @@
{ "name": "ClassicToDisklessStartOffset", "type": "int64", "versions": "0+",
"about": "The committed seal: the first offset in diskless format." },
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The leader epoch seen by the broker." }
"about": "The leader epoch seen by the broker." },
{ "name": "LastClassicLeaderEpoch", "type": "int32", "versions": "0+", "taggedVersions": "0+", "tag": 0,
"default": "-1",
"about": "The leader epoch of the record at the seal minus one, or -1 if the leader holds no classic record below the seal." }
]}
]}
]
Expand Down
60 changes: 60 additions & 0 deletions core/src/main/java/kafka/server/TierStateMachine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData;
10BC8 import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.metadata.PartitionRegistration;
import org.apache.kafka.server.LeaderEndPoint;
import org.apache.kafka.server.PartitionFetchState;
import org.apache.kafka.server.ReplicaState;
Expand Down Expand Up @@ -198,6 +199,15 @@ private Long buildRemoteLogAuxState(TopicPartition topicPartition,

RemoteLogManager rlm = replicaMgr.remoteLogManager().get();

long seal = replicaMgr.classicToDisklessStartOffset(topicPartition);
if (seal > 0 && leaderLocalLogStartOffset <= seal) {
int lastClassicLeaderEpoch = replicaMgr.lastClassicLeaderEpoch(topicPartition);
if (lastClassicLeaderEpoch != PartitionRegistration.NO_LAST_CLASSIC_LEADER_EPOCH) {
return buildClassicPrefixAuxState(topicPartition, currentLeaderEpoch, leaderLocalLogStartOffset,
leaderLogStartOffset, seal, lastClassicLeaderEpoch, rlm, unifiedLog);
}
}

// Find the respective leader epoch for (leaderLocalLogStartOffset - 1). We need to build the leader epoch cache
// until that offset
long previousOffsetToLeaderLocalLogStartOffset = leaderLocalLogStartOffset - 1;
Expand Down Expand Up @@ -265,6 +275,56 @@ private Long buildRemoteLogAuxState(TopicPartition topicPartition,
return nextOffset;
}

/**
* Inkless: builds the state below the classic-to-diskless seal in phase 2 of the switch and returns the seal as the
* next offset to fetch.
* <p>
* Remote storage can hold a segment from a leader epoch that an unclean election discarded. Such a segment can
* hold the record before the local log start of the leader in the same epoch as the committed history and diverge
* after it. Each leader epoch has one leader, so a segment that holds {@code seal - 1} in the last classic leader
* epoch holds the committed history below the seal, even if it also holds discarded records past the seal. The
* local log starts at the seal because the replica doesn't keep local classic records in phase 2.
*/
private long buildClassicPrefixAuxState(TopicPartition topicPartition,
int currentLeaderEpoch,
long leaderLocalLogStartOffset,
long leaderLogStartOffset,
long seal,
int lastClassicLeaderEpoch,
RemoteLogManager rlm,
UnifiedLog unifiedLog) throws IOException, RemoteStorageException {
if (!rlm.isPartitionReady(topicPartition)) {
throw new RemoteStorageNotReadyException("RemoteLogManager is not ready for partition: " + topicPartition);
}

RemoteLogSegmentMetadata remoteLogSegmentMetadata = rlm.fetchRemoteLogSegmentMetadata(topicPartition, lastClassicLeaderEpoch, seal - 1)
.orElseThrow(() -> buildRemoteStorageException(topicPartition, lastClassicLeaderEpoch, currentLeaderEpoch,
leaderLocalLogStartOffset, leaderLogStartOffset));

Partition partition = replicaMgr.getPartitionOrException(topicPartition);
partition.truncateFullyAndStartAt(seal, useFutureLog, Optional.of(leaderLogStartOffset));
unifiedLog.maybeIncrementLogStartOffset(leaderLogStartOffset, LeaderOffsetIncremented);
unifiedLog.maybeIncrementLocalLogStartOffset(seal, LeaderOffsetIncremented);

List<EpochEntry> epochs = readLeaderEpochCheckpoint(rlm, remoteLogSegmentMetadata).stream()
.filter(entry -> entry.startOffset() < seal)
.toList();
unifiedLog.leaderEpochCache().assign(epochs);

// The producer snapshot of a segment is at its end. Sealing aborts undecided transactions, so the empty
// producer state that the truncation leaves is correct at the seal.
if (remoteLogSegmentMetadata.endOffset() + 1 == seal) {
buildProducerSnapshotFile(unifiedLog, seal, remoteLogSegmentMetadata, rlm);
} else {
log.warn("Remote segment {} for {} ends at {}, past the classic-to-diskless seal {}; rebuilding up to the seal",
remoteLogSegmentMetadata.remoteLogSegmentId(), topicPartition, remoteLogSegmentMetadata.endOffset(), seal);
}

log.info("Rebuilt the state of {} below the classic-to-diskless seal {} from remote segment {} with {} leader epochs",
topicPartition, seal, remoteLogSegmentMetadata.remoteLogSegmentId(), epochs.size());
return seal;
}

private RemoteStorageException buildRemoteStorageException(TopicPartition topicPartition,
int targetEpoch,
int currentLeaderEpoch,
Expand Down
36 changes: 31 additions & 5 deletions core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,7 @@ class DisklessLeaderEndPoint(
fetchResponseData.setLogStartOffset(UnifiedLog.UNKNOWN_OFFSET)
case Right(partition) =>
val localLogOpt = Try(partition.localLogOrException).toOption
val crossTier = replicaManager.crossTierEarliestOffset(tp.topicPartition)
val logStartOffset = localLogOpt match {
case Some(localLog) =>
// Prefer the control-plane cross-tier earliest as the whole-log start for a consolidating
Expand All @@ -154,7 +155,6 @@ class DisklessLeaderEndPoint(
// OFFSET_MOVED_TO_TIERED_STORAGE gate below requires requestedOffset >= logStartOffset. The
// cross-tier earliest is broker-agnostic and authoritative, and min() never advances past data
// the local log still holds, so it is the safe whole-log start.
val crossTier = replicaManager.crossTierEarliestOffset(tp.topicPartition)
if (crossTier.isPresent) Math.min(crossTier.getAsLong, localLog.logStartOffset)
else localLog.logStartOffset
case None =>
Expand All @@ -180,16 +180,28 @@ class DisklessLeaderEndPoint(
// or a covering segment still transitional) returns NOT_LEADER_OR_FOLLOWER so the
// fetcher retries at debug. Fetchers start in applyLocalLeadersDelta before
// rlm.onLeadershipChange, and isReady can stay false until __remote_log_metadata
// catch-up.
val disklessStartOffset = data.logStartOffset
// catch-up. A switched partition's WAL starts at its classic log start until the
// first prune moves it to the seal. The seal is the lowest diskless start.
val walStartOffset = data.logStartOffset
val disklessStartOffset = math.max(walStartOffset, replicaManager.classicToDisklessStartOffset(tp.topicPartition))
val requestedOffset = Option(fetchInfos.get(tp)).map(_.fetchOffset).getOrElse(-1L)
if (logStartOffset != UnifiedLog.UNKNOWN_OFFSET &&
requestedOffset >= logStartOffset &&
requestedOffset < disklessStartOffset) {
localLogOpt.filter(_.remoteLogEnabled()).foreach { _ =>
val generation = remotePrefixGenerations.getOrElse(tp.topicPartition, 0L)
val evidence = remoteConsolidatedPrefixEvidence(tp.topicPartition, disklessStartOffset)
if (!evidence.isPresent) {
lazy val evidence = remoteConsolidatedPrefixEvidence(tp.topicPartition, disklessStartOffset)
if (crossTier.isPresent && crossTier.getAsLong >= disklessStartOffset) {
// Retention or DeleteRecords removed everything below the diskless start. The
// out-of-range reset starts the log at the cross-tier earliest, so the fetcher
// neither rebuilds a deleted prefix nor waits for remote coverage that never comes.
replicaManager.markConsolidationRemotePrefixUnknown(tp.topicPartition, unknown = false, generation)
logger.debug("Offset {} for {} is below the cross-tier earliest {}, which is at or past the diskless start {}; " +
"returning OFFSET_OUT_OF_RANGE so the fetcher resets to the cross-tier earliest.",
requestedOffset, tp.topicPartition, crossTier.getAsLong, disklessStartOffset)
fetchResponseData.setErrorCode(Errors.OFFSET_OUT_OF_RANGE.code)
fetchResponseData.setRecords(MemoryRecords.EMPTY)
} else if (!evidence.isPresent) {
// leader.fetch() runs outside partitionMapLock. Alert semantics: the
// ConsolidationRemotePrefixUnknown row in docs/inkless/DISKLESS_CONSOLIDATION.md.
replicaManager.markConsolidationRemotePrefixUnknown(tp.topicPartition, unknown = true, generation)
Expand All @@ -206,6 +218,16 @@ class DisklessLeaderEndPoint(
replicaManager.recordConsolidationTierRebuildRequest(tp.topicPartition)
fetchResponseData.setErrorCode(Errors.OFFSET_MOVED_TO_TIERED_STORAGE.code)
fetchResponseData.setRecords(MemoryRecords.EMPTY)
} else if (requestedOffset >= walStartOffset) {
// Only the seal puts the offset below the diskless start. The WAL returns the first
// diskless batch for it, which would hide the classic prefix. An out-of-range reset
// fetches the same offset again without a backoff.
replicaManager.markConsolidationRemotePrefixUnknown(tp.topicPartition, unknown = true, generation)
Comment thread
giuseppelillo marked this conversation as resolved.
Resolved
logger.debug("RLMM has no readable coverage of {} for {}, below the classic-to-diskless seal; " +
"returning NOT_LEADER_OR_FOLLOWER so the fetcher retries instead of consolidating below the seal at offset {}.",
disklessStartOffset - 1, tp.topicPartition, requestedOffset)
fetchResponseData.setErrorCode(Errors.NOT_LEADER_OR_FOLLOWER.code)
fetchResponseData.setRecords(MemoryRecords.EMPTY)
} else {
replicaManager.markConsolidationRemotePrefixUnknown(tp.topicPartition, unknown = false, generation)
}
Expand Down Expand Up @@ -355,6 +377,10 @@ class DisklessLeaderEndPoint(
val offset =
if (timestamp == ListOffsetsRequest.LATEST_TIMESTAMP)
atLeastCommittedSeal(topicPartition, tao.offset)
else if (timestamp == ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP)
// The tier-state rebuild starts the local log here. Before the first prune of a switched
// partition, the WAL start is still the classic log start, which remote storage holds.
math.max(tao.offset, replicaManager.classicToDisklessStartOffset(topicPartition))
else
tao.offset
val taoForEpoch =
Expand Down
15 changes: 14 additions & 1 deletion core/src/main/scala/kafka/server/InitDisklessLogState.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,13 @@ import kafka.utils.Logging
import org.apache.kafka.common.{TopicPartition, Uuid}
import org.apache.kafka.common.message.{InitDisklessLogRequestData, InitDisklessLogResponseData, SetClassicPrefixInRemoteStorageRequestData, SetClassicPrefixInRemoteStorageResponseData}
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.metadata.PartitionRegistration
import org.apache.kafka.server.partition.PartitionListener
import org.apache.kafka.storage.internals.log.UnifiedLog

import java.util
import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters._

sealed trait InitDisklessLogState extends Logging {
def partition: Partition
Expand Down Expand Up @@ -308,7 +310,8 @@ object SendingFlagToController {
.add(new SetClassicPrefixInRemoteStorageRequestData.PartitionData()
.setPartitionId(state.tp.partition)
.setClassicToDisklessStartOffset(state.seal)
.setLeaderEpoch(state.partition.getLeaderEpoch))
.setLeaderEpoch(state.partition.getLeaderEpoch)
.setLastClassicLeaderEpoch(lastClassicLeaderEpoch(state)))
}

val topicDataList = new util.ArrayList[SetClassicPrefixInRemoteStorageRequestData.TopicData]()
Expand All @@ -324,6 +327,16 @@ object SendingFlagToController {
.setTopics(topicDataList)
}

/**
* Returns the leader epoch of the record at `seal - 1`, or `NO_LAST_CLASSIC_LEADER_EPOCH` if the log doesn't
* contain a classic record below the seal.
*/
private def lastClassicLeaderEpoch(state: SendingFlagToController): Int =
state.partition.log
.filter(log => log.logStartOffset < state.seal)
.flatMap(log => log.leaderEpochCache.epochForOffset(state.seal - 1).toScala)
.getOrElse(PartitionRegistration.NO_LAST_CLASSIC_LEADER_EPOCH)

def parseBatchResponse(response: SetClassicPrefixInRemoteStorageResponseData): Iterable[ParsedResponse] = {
val outcomes = scala.collection.mutable.ArrayBuffer[ParsedResponse]()
for (topicResponse <- response.topics().asScala) {
Expand Down
Loading
Loading
0