You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dybyte (via GitHub)" <gi...@apache.org> on 2026/09/23 19:50:45 UTC
[PR] KAFKA-17249: Do not truncate the local log before the remote log aux state is fetched [kafka]
dybyte opened a new pull request, #23569:
URL: https://github.com/apache/kafka/pull/23569
When building the remote log aux state, TierStateMachine truncates the local log, which also clears the leader epoch cache on disk, before it reads the leader epoch checkpoint and the producer snapshot from remote storage. If a remote read fails, the replica claims the remote range [logStartOffset, localLogStartOffset) with no leader epoch entries for it.
The replica does not recover from this after a restart or a leader change. The fetcher then starts from the end offset of the truncated log, which is inside the leader's local log, so OFFSET_MOVED_TO_TIERED_STORAGE is never returned again and the aux state is never rebuilt. The replica rejoins the ISR. If it then becomes the leader, it cannot serve the remote range, and its remote log retention deletes the remote segments of the lost epochs as unreferenced. This is the path in the reproduction steps of the JIRA.
This change reads both files from remote storage into memory first and modifies the local log only after both reads succeed. A remote failure now leaves the local log unchanged, so the next fetch retries the aux state build.
Testing: the new TierStateMachineBuildRemoteLogAuxStateTest makes the remote leader epoch read fail and checks that the log start offsets and the leader epoch cache, in memory and on disk, are unchanged. It fails on trunk and passes with this change. TierStateMachineTest and the tiered storage integration tests AlterLogDirTest, ReassignReplicaMoveAndExpandTest and FetchFromLeaderWithCorruptedCheckpointTest also pass.
Co-Authored-By: Claude
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17249: Do not truncate the local log before the remote log aux state is fetched [kafka]
Posted by "dybyte (via GitHub)" <gi...@apache.org>.
dybyte commented on code in PR #23569:
URL: https://github.com/apache/kafka/pull/23569#discussion_r4103150574
##########
core/src/main/java/kafka/server/TierStateMachine.java:
##########
@@ -243,20 +246,24 @@ private Long buildRemoteLogAuxState(TopicPartition topicPartition,
// Assign nextOffset with the offset from which next fetch should happen.
long nextOffset = remoteLogSegmentMetadata.endOffset() + 1;
- // Truncate the existing local log before restoring the leader epoch cache and producer snapshots.
+ // Read everything needed from remote storage before modifying the local log, so that a failed read
+ // leaves the local log and its leader epoch cache untouched and the next fetch retries this state machine.
+ List<EpochEntry> epochs = readLeaderEpochCheckpoint(rlm, remoteLogSegmentMetadata);
Review Comment:
Thanks, sounds good to me.
1. Added. When a partition becomes the leader, RemoteLogManager logs a warning if the log has a remote range (logStartOffset < localLogStartOffset) and the epoch cache is empty or starts after logStartOffset. The check is a separate method so 2 and 3 can reuse it.
2. +1. I verified with a local test that the unreferenced-epoch cleanup deletes all remote segments of the lost epochs in this state. Such a leader also uploads its incomplete checkpoint with every new segment, so followers that later rebuild from it inherit the gap. What do you think about also skipping the copy while the cache is incomplete, using the same check? The cost is that local segments cannot be deleted until they are copied, so disk usage grows until leadership moves to a healthy replica. Is that a reasonable direction, or is the warning enough for this case?
3. +1, but I'd truncate the local log to logStartOffset before refetching. After a restart the follower's logStartOffset can lag the leader's, and then OFFSET_OUT_OF_RANGE makes the fetcher resume from the local log end offset, skipping the rebuild.
Neither 2 nor 3 fixes a replica that is already the leader. I'll open a follow-up JIRA for 2 and 3 and mention that there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17249: Do not truncate the local log before the remote log aux state is fetched [kafka]
Posted by "dybyte (via GitHub)" <gi...@apache.org>.
dybyte commented on code in PR #23569:
URL: https://github.com/apache/kafka/pull/23569#discussion_r4148184373
##########
core/src/main/java/kafka/server/TierStateMachine.java:
##########
@@ -243,20 +246,24 @@ private Long buildRemoteLogAuxState(TopicPartition topicPartition,
// Assign nextOffset with the offset from which next fetch should happen.
long nextOffset = remoteLogSegmentMetadata.endOffset() + 1;
- // Truncate the existing local log before restoring the leader epoch cache and producer snapshots.
+ // Read everything needed from remote storage before modifying the local log, so that a failed read
+ // leaves the local log and its leader epoch cache untouched and the next fetch retries this state machine.
+ List<EpochEntry> epochs = readLeaderEpochCheckpoint(rlm, remoteLogSegmentMetadata);
Review Comment:
I opened KAFKA-21192 for 2 and KAFKA-21191 for 3. The open questions are in KAFKA-21192.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17249: Do not truncate the local log before the remote log aux state is fetched [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23569:
URL: https://github.com/apache/kafka/pull/23569#discussion_r4121268229
##########
core/src/test/scala/kafka/server/TierStateMachineBuildRemoteLogAuxStateTest.scala:
##########
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import kafka.utils.{Logging, TestUtils}
+
+import org.apache.kafka.common.compress.Compression
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.metadata.{FeatureLevelRecord, PartitionRecord, TopicRecord}
+import org.apache.kafka.common.metrics.Metrics
+import org.apache.kafka.common.record.internal.{MemoryRecords, SimpleRecord}
+import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.{TopicIdPartition, Uuid}
+import org.apache.kafka.image.{MetadataDelta, MetadataImage, MetadataProvenance}
+import org.apache.kafka.metadata.KRaftMetadataCache
+import org.apache.kafka.server.LeaderEndPoint
+import org.apache.kafka.server.common.{KRaftVersion, MetadataVersion, OffsetAndEpoch}
+import org.apache.kafka.server.log.remote.storage.{RemoteLogManager, RemoteLogSegmentMetadata, RemoteStorageException, RemoteStorageManager}
+import org.apache.kafka.server.log.remote.storage.RemoteStorageManager.IndexType
+import org.apache.kafka.server.partition.AlterPartitionManager
+import org.apache.kafka.server.quota.QuotaFactory
+import org.apache.kafka.server.quota.QuotaFactory.QuotaManagers
+import org.apache.kafka.server.util.{MockScheduler, MockTime}
+import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile
+import org.apache.kafka.storage.internals.log.{EpochEntry, LogConfig, LogDirFailureChannel, UnifiedLog}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.{AfterEach, BeforeEach}
+import org.junit.jupiter.api.function.Executable
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.EnumSource
+import org.mockito.ArgumentMatchers.{any, anyInt, anyLong}
+import org.mockito.Mockito.{mock, when}
+
+import java.io.{ByteArrayInputStream, File}
+import java.nio.charset.StandardCharsets
+import java.util.{Optional, Properties}
+import scala.collection.Map
+import scala.jdk.CollectionConverters._
+
+/**
+ * Tests that a failure while building the remote log auxiliary state does not leave the follower's
+ * local log and its leader epoch cache in an inconsistent state (KAFKA-17249).
+ */
+class TierStateMachineBuildRemoteLogAuxStateTest extends Logging {
Review Comment:
Would you mind using Java to rewrite this test? We are dropping Scala code, so it would be better to use Java as much as possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17249: Do not truncate the local log before the remote log aux state is fetched [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23569:
URL: https://github.com/apache/kafka/pull/23569#discussion_r4221709148
##########
core/src/test/java/kafka/server/TierStateMachineBuildRemoteLogAuxStateTest.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server;
+
+import kafka.server.builders.LogManagerBuilder;
Review Comment:
Please rebase code. `LogManagerBuilder` is removed already
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -498,10 +498,36 @@ public void onLeadershipChange(Set<TopicPartitionLog> partitionsBecomeLeader,
// background thread and might emit metrics. So, removing the metrics after marking this node as follower.
followerPartitions.forEach((tp, __) -> removeRemoteTopicPartitionMetrics(tp));
+ filterPartitions(partitionsBecomeLeader)
+ .forEach(partition -> warnIfLeaderEpochCacheMissesRemoteRange(partition.unifiedLog().get()));
leaderPartitions.forEach(this::doHandleLeaderPartition);
}
}
+ private void warnIfLeaderEpochCacheMissesRemoteRange(UnifiedLog log) {
+ if (isLeaderEpochCacheMissingRemoteEntries(log)) {
+ LOGGER.warn("The leader epoch cache of {} does not cover the remote log range [{}, {}), its earliest entry is {}. " +
+ "Remote log segments of the missing epochs cannot be read through this replica and may be " +
+ "deleted as unreferenced.",
+ log.topicPartition(), log.logStartOffset(), log.localLogStartOffset(),
+ log.leaderEpochCache().earliestEntry().orElse(null));
+ }
+ }
+
+ /**
+ * Returns true if the log has a remote range [logStartOffset, localLogStartOffset) whose leader epochs are
+ * missing from the leader epoch cache, which happens when the remote log aux state was lost (KAFKA-17249).
+ */
+ // Visible for testing
+ static boolean isLeaderEpochCacheMissingRemoteEntries(UnifiedLog log) {
+ long logStartOffset = log.logStartOffset();
+ if (logStartOffset >= log.localLogStartOffset()) {
Review Comment:
The case of “=” is what we really care about, right? If the start offset is larger than the local start offset, that is a real problem and we should not silently treat it as “no remote range” :) Not a blocker, just thinking out loud.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17249: Do not truncate the local log before the remote log aux state is fetched [kafka]
Posted by "dybyte (via GitHub)" <gi...@apache.org>.
dybyte commented on code in PR #23569:
URL: https://github.com/apache/kafka/pull/23569#discussion_r4261685691
##########
core/src/test/java/kafka/server/TierStateMachineBuildRemoteLogAuxStateTest.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server;
+
+import kafka.server.builders.LogManagerBuilder;
Review Comment:
Done, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17249: Do not truncate the local log before the remote log aux state is fetched [kafka]
Posted by "dybyte (via GitHub)" <gi...@apache.org>.
dybyte commented on code in PR #23569:
URL: https://github.com/apache/kafka/pull/23569#discussion_r4124608287
##########
core/src/test/scala/kafka/server/TierStateMachineBuildRemoteLogAuxStateTest.scala:
##########
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import kafka.utils.{Logging, TestUtils}
+
+import org.apache.kafka.common.compress.Compression
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.metadata.{FeatureLevelRecord, PartitionRecord, TopicRecord}
+import org.apache.kafka.common.metrics.Metrics
+import org.apache.kafka.common.record.internal.{MemoryRecords, SimpleRecord}
+import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.{TopicIdPartition, Uuid}
+import org.apache.kafka.image.{MetadataDelta, MetadataImage, MetadataProvenance}
+import org.apache.kafka.metadata.KRaftMetadataCache
+import org.apache.kafka.server.LeaderEndPoint
+import org.apache.kafka.server.common.{KRaftVersion, MetadataVersion, OffsetAndEpoch}
+import org.apache.kafka.server.log.remote.storage.{RemoteLogManager, RemoteLogSegmentMetadata, RemoteStorageException, RemoteStorageManager}
+import org.apache.kafka.server.log.remote.storage.RemoteStorageManager.IndexType
+import org.apache.kafka.server.partition.AlterPartitionManager
+import org.apache.kafka.server.quota.QuotaFactory
+import org.apache.kafka.server.quota.QuotaFactory.QuotaManagers
+import org.apache.kafka.server.util.{MockScheduler, MockTime}
+import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile
+import org.apache.kafka.storage.internals.log.{EpochEntry, LogConfig, LogDirFailureChannel, UnifiedLog}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.{AfterEach, BeforeEach}
+import org.junit.jupiter.api.function.Executable
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.EnumSource
+import org.mockito.ArgumentMatchers.{any, anyInt, anyLong}
+import org.mockito.Mockito.{mock, when}
+
+import java.io.{ByteArrayInputStream, File}
+import java.nio.charset.StandardCharsets
+import java.util.{Optional, Properties}
+import scala.collection.Map
+import scala.jdk.CollectionConverters._
+
+/**
+ * Tests that a failure while building the remote log auxiliary state does not leave the follower's
+ * local log and its leader epoch cache in an inconsistent state (KAFKA-17249).
+ */
+class TierStateMachineBuildRemoteLogAuxStateTest extends Logging {
Review Comment:
Done, thanks for the review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17249: Do not truncate the local log before the remote log aux state is fetched [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23569:
URL: https://github.com/apache/kafka/pull/23569#discussion_r4100224664
##########
core/src/main/java/kafka/server/TierStateMachine.java:
##########
@@ -243,20 +246,24 @@ private Long buildRemoteLogAuxState(TopicPartition topicPartition,
// Assign nextOffset with the offset from which next fetch should happen.
long nextOffset = remoteLogSegmentMetadata.endOffset() + 1;
- // Truncate the existing local log before restoring the leader epoch cache and producer snapshots.
+ // Read everything needed from remote storage before modifying the local log, so that a failed read
+ // leaves the local log and its leader epoch cache untouched and the next fetch retries this state machine.
+ List<EpochEntry> epochs = readLeaderEpochCheckpoint(rlm, remoteLogSegmentMetadata);
Review Comment:
I'm thinking how to fix the epoch files which have encountered this bug.
1. If the earliest entry of the leader epoch cache has a start offset larger than the log start offset, we should add a warning message when the follower becomes a leader. This can be addressed in this PR.
2. We should consider skipping the deletion of remote segments if the epoch cache is incomplete.
3. If this bug happens on a follower only, it can rebuild the whole cache by resetting the fetch offset to the log start offset.
Steps 2 and 3 could be the follow-ups. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17249: Do not truncate the local log before the remote log aux state is fetched [kafka]
Posted by "dybyte (via GitHub)" <gi...@apache.org>.
dybyte commented on code in PR #23569:
URL: https://github.com/apache/kafka/pull/23569#discussion_r4261672079
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -498,10 +498,36 @@ public void onLeadershipChange(Set<TopicPartitionLog> partitionsBecomeLeader,
// background thread and might emit metrics. So, removing the metrics after marking this node as follower.
followerPartitions.forEach((tp, __) -> removeRemoteTopicPartitionMetrics(tp));
+ filterPartitions(partitionsBecomeLeader)
+ .forEach(partition -> warnIfLeaderEpochCacheMissesRemoteRange(partition.unifiedLog().get()));
leaderPartitions.forEach(this::doHandleLeaderPartition);
}
}
+ private void warnIfLeaderEpochCacheMissesRemoteRange(UnifiedLog log) {
+ if (isLeaderEpochCacheMissingRemoteEntries(log)) {
+ LOGGER.warn("The leader epoch cache of {} does not cover the remote log range [{}, {}), its earliest entry is {}. " +
+ "Remote log segments of the missing epochs cannot be read through this replica and may be " +
+ "deleted as unreferenced.",
+ log.topicPartition(), log.logStartOffset(), log.localLogStartOffset(),
+ log.leaderEpochCache().earliestEntry().orElse(null));
+ }
+ }
+
+ /**
+ * Returns true if the log has a remote range [logStartOffset, localLogStartOffset) whose leader epochs are
+ * missing from the leader epoch cache, which happens when the remote log aux state was lost (KAFKA-17249).
+ */
+ // Visible for testing
+ static boolean isLeaderEpochCacheMissingRemoteEntries(UnifiedLog log) {
+ long logStartOffset = log.logStartOffset();
+ if (logStartOffset >= log.localLogStartOffset()) {
Review Comment:
Good question. `>` can actually happen in a normal flow. While remote storage is disabled for a topic, `maybeIncrementLogStartOffset` doesn't move `localLogStartOffset`, so retention or DeleteRecords leaves `logStartOffset > localLogStartOffset`. If remote storage is then enabled on that topic, `ConfigHandler` calls `onLeadershipChange` and this check runs in exactly that state. There is no remote data yet at that point, so "no remote range" is the right answer, and flagging it would log a false warning during a normal operation like enabling tiered storage on an existing topic.
The stale `localLogStartOffset` itself is a separate bug, since it makes ListOffsets return wrong values for EARLIEST_LOCAL and EARLIEST_PENDING_UPLOAD. I filed KAFKA-21272 for it. It doesn't overlap with this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org