You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "abhijeetk88 (via GitHub)" <gi...@apache.org> on 2025/08/28 07:05:55 UTC
[PR] [WIP] KAFKA-17302 [kafka]
abhijeetk88 opened a new pull request, #20428:
URL: https://github.com/apache/kafka/pull/20428
Delete this text and replace it with a detailed description of your change. The
PR title and body will become the squashed commit message.
If you would like to tag individuals, add some commentary, upload images, or
include other supplemental information that should not be part of the eventual
commit message, please use a separate comment.
If applicable, please include a summary of the testing strategy (including
rationale) for the proposed change. Unit and/or integration tests are expected
for any behavior change and system tests should be considered for larger
changes.
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #20428:
URL: https://github.com/apache/kafka/pull/20428#issuecomment-3647587385
@abhijeetk88
Could you rebase the PR and resolve the open comments? 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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph merged PR #20428:
URL: https://github.com/apache/kafka/pull/20428
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #20428:
URL: https://github.com/apache/kafka/pull/20428#issuecomment-3811415961
@abhijeetk88
Could you please split this patch into two?
The below case seem to bring additional complexity to the code and can be taken in the next patch if required:
```
If this code change is to handle the case when leader-log-start-offset == leader-local-log-start-offset == 0 and earliestUploadOffset > 0 (yet to be approved), then we may have to discuss it before proceeding with the patch. It is one of the edge-case but not sure how useful it would be in practice.
```
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on PR #20428:
URL: https://github.com/apache/kafka/pull/20428#issuecomment-3721111020
@abhijeetk88 I saw PR: https://github.com/apache/kafka/pull/21153/files?diff=split&w=0 is merged, are you going to come back to this PR to finish it up?
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on PR #20428:
URL: https://github.com/apache/kafka/pull/20428#issuecomment-3729879039
@HenryCaiHaiying, working to close this out.
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "songpinru (via GitHub)" <gi...@apache.org>.
songpinru commented on PR #20428:
URL: https://github.com/apache/kafka/pull/20428#issuecomment-3588172187
I hava an question:
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #20428:
URL: https://github.com/apache/kafka/pull/20428#discussion_r2317900129
##########
core/src/main/scala/kafka/server/LocalLeaderEndPoint.scala:
##########
@@ -135,6 +135,35 @@ class LocalLeaderEndPoint(sourceBroker: BrokerEndPoint,
new OffsetAndEpoch(localLogStartOffset, epoch.orElse(0))
}
+ override def fetchEarliestPendingUploadOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int): OffsetAndEpoch = {
+ val partition = replicaManager.getPartitionOrException(topicPartition)
+ val log = partition.localLogOrException
+
+ if (!log.remoteLogEnabled())
+ return new OffsetAndEpoch(-1L, -1)
+
+ val highestRemoteOffset = log.highestOffsetInRemoteStorage()
+
+ if (highestRemoteOffset == -1L) {
+ val localLogStartOffset = fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch)
+ val logStartOffset = fetchEarliestOffset(topicPartition, currentLeaderEpoch)
+
+ if (localLogStartOffset.offset() == logStartOffset.offset()) {
+ // No segments have been uploaded yet
+ return logStartOffset;
+ } else {
+ // Leader currently does not know about the already uploaded segments
+ return new OffsetAndEpoch(-1L, -1);
+ }
+ }
+
+ val logStartOffset = fetchEarliestOffset(topicPartition, currentLeaderEpoch)
+ val earliestPendingUploadOffset = Math.max(highestRemoteOffset + 1, logStartOffset.offset())
+ val epoch = log.leaderEpochCache.epochForOffset(earliestPendingUploadOffset)
+
+ new OffsetAndEpoch(earliestPendingUploadOffset, epoch.orElse(0))
Review Comment:
epoch 0 is vaild. If the epoch is empty, shall we return -1?
##########
core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala:
##########
@@ -818,4 +820,96 @@ class ReplicaFetcherThreadTest {
when(replicaManager.localLogOrException(t2p1)).thenReturn(log)
when(replicaManager.getPartitionOrException(t2p1)).thenReturn(partition)
}
+
+ @ParameterizedTest
+ @CsvSource(Array(
+ "false, false, compact, 0, 0, false",
+ "false, false, compact, 5, 0, false",
+ "false, false, compact, 5, 1, false",
+ "false, false, delete, 0, 0, false",
+ "false, false, delete, 5, 0, false",
+ "false, false, delete, 5, 1, false",
+ "false, true, compact, 0, 0, false",
+ "false, true, compact, 5, 0, false",
+ "false, true, compact, 5, 1, false",
+ "false, true, delete, 0, 0, false",
+ "false, true, delete, 5, 0, false",
+ "false, true, delete, 5, 1, false",
+ "true, false, compact, 0, 0, false",
+ "true, false, compact, 5, 0, false",
+ "true, false, compact, 5, 1, false",
+ "true, false, delete, 0, 0, false",
+ "true, false, delete, 5, 0, false",
+ "true, false, delete, 5, 1, false",
+ "true, true, compact, 0, 0, false",
+ "true, true, compact, 5, 0, false",
+ "true, true, compact, 5, 1, false",
+ "true, true, delete, 0, 0, false",
+ "true, true, delete, 5, 0, true",
+ "true, true, delete, 5, 1, false"))
+ def testShouldFetchFromLastTieredOffset(enableLastTieredOffsetFetch: Boolean,
+ remoteStorageEnabled: Boolean,
+ cleanUpPolicy: String,
+ leaderEndOffset: Long,
+ replicaEndOffset: Long,
+ expected: Boolean): Unit = {
+ val tp = new TopicPartition("t", 0)
+
+ def runScenario(enableLastTieredOffsetFetch: Boolean,
Review Comment:
do we need this internal `runScenario` method? Can this be flattened?
```java
assertEquals(
expected,
thread.callShouldFetchFromLastTieredOffset(tp, leaderEndOffset, replicaEndOffset)
)
```
##########
core/src/main/java/kafka/server/TierStateMachine.java:
##########
@@ -182,7 +179,7 @@ private void buildProducerSnapshotFile(UnifiedLog unifiedLog,
* fetching records from the leader. The return value is the next offset to fetch from the leader, which is the
* next offset following the end offset of the remote log portion.
*/
- private Long buildRemoteLogAuxState(TopicPartition topicPartition,
+ protected Long buildRemoteLogAuxState(TopicPartition topicPartition,
Review Comment:
nit: fix alignment
##########
core/src/test/scala/unit/kafka/server/MockLeaderEndPoint.scala:
##########
@@ -135,6 +135,16 @@ class MockLeaderEndPoint(sourceBroker: BrokerEndPoint = new BrokerEndPoint(1, "l
new OffsetAndEpoch(leaderState.localLogStartOffset, leaderState.leaderEpoch)
}
+ override def fetchEarliestPendingUploadOffset(topicPartition: TopicPartition, leaderEpoch: Int): OffsetAndEpoch = {
+ val leaderState = leaderPartitionState(topicPartition)
+ checkLeaderEpochAndThrow(leaderEpoch, leaderState)
+ val offsetAndEpoch = leaderState.earliestPendingUploadOffset match {
+ case -1L => (-1L, -1)
Review Comment:
can the OffsetAndEpoch be returned directly?
```
leaderState.earliestPendingUploadOffset match {
case -1L => new OffsetAndEpoch(-1L, -1)
case _ => new OffsetAndEpoch(math.max(leaderState.earliestPendingUploadOffset, leaderState.logStartOffset), leaderState.leaderEpoch)
}
```
##########
core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala:
##########
@@ -818,4 +820,96 @@ class ReplicaFetcherThreadTest {
when(replicaManager.localLogOrException(t2p1)).thenReturn(log)
when(replicaManager.getPartitionOrException(t2p1)).thenReturn(partition)
}
+
+ @ParameterizedTest
+ @CsvSource(Array(
+ "false, false, compact, 0, 0, false",
+ "false, false, compact, 5, 0, false",
+ "false, false, compact, 5, 1, false",
+ "false, false, delete, 0, 0, false",
+ "false, false, delete, 5, 0, false",
+ "false, false, delete, 5, 1, false",
+ "false, true, compact, 0, 0, false",
+ "false, true, compact, 5, 0, false",
+ "false, true, compact, 5, 1, false",
+ "false, true, delete, 0, 0, false",
+ "false, true, delete, 5, 0, false",
+ "false, true, delete, 5, 1, false",
+ "true, false, compact, 0, 0, false",
+ "true, false, compact, 5, 0, false",
+ "true, false, compact, 5, 1, false",
+ "true, false, delete, 0, 0, false",
+ "true, false, delete, 5, 0, false",
+ "true, false, delete, 5, 1, false",
+ "true, true, compact, 0, 0, false",
+ "true, true, compact, 5, 0, false",
+ "true, true, compact, 5, 1, false",
+ "true, true, delete, 0, 0, false",
+ "true, true, delete, 5, 0, true",
+ "true, true, delete, 5, 1, false"))
+ def testShouldFetchFromLastTieredOffset(enableLastTieredOffsetFetch: Boolean,
+ remoteStorageEnabled: Boolean,
+ cleanUpPolicy: String,
+ leaderEndOffset: Long,
+ replicaEndOffset: Long,
+ expected: Boolean): Unit = {
+ val tp = new TopicPartition("t", 0)
+
+ def runScenario(enableLastTieredOffsetFetch: Boolean,
+ cleanupPolicy: String,
+ remoteStorageEnabled: Boolean,
+ leaderEndOffset: Long,
+ replicaEndOffset: Long): Boolean = {
+ val props = TestUtils.createBrokerConfig(1)
+ props.put(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG, String.valueOf(enableLastTieredOffsetFetch))
+ val config = KafkaConfig.fromProps(props)
+
+ val mockBlockingSend: BlockingSend = mock(classOf[BlockingSend])
+ when(mockBlockingSend.brokerEndPoint()).thenReturn(brokerEndPoint)
+
+ val replicaManager: ReplicaManager = mock(classOf[ReplicaManager])
+ when(replicaManager.brokerTopicStats).thenReturn(new BrokerTopicStats)
+
+ val lcOverrides = new Properties()
+ lcOverrides.put(TopicConfig.CLEANUP_POLICY_CONFIG, cleanupPolicy)
+ val logConfig = LogConfig.fromProps(config.extractLogConfigMap, lcOverrides)
+
+ val log: UnifiedLog = mock(classOf[UnifiedLog])
+ when(log.config).thenReturn(logConfig)
+ when(log.remoteLogEnabled()).thenReturn(remoteStorageEnabled)
+ when(replicaManager.localLog(tp)).thenReturn(Some(log))
+
+ val logContext = new LogContext(s"[ReplicaFetcher replicaId=${config.brokerId}, leaderId=${mockBlockingSend.brokerEndPoint().id}, fetcherId=0] ")
+ val fetchSessionHandler = new FetchSessionHandler(logContext, mockBlockingSend.brokerEndPoint().id)
+ val leader = new RemoteLeaderEndPoint(logContext.logPrefix, mockBlockingSend, fetchSessionHandler,
+ config, replicaManager, UNBOUNDED_QUOTA, () => MetadataVersion.MINIMUM_VERSION, () => 1)
+
+ val thread = new MockReplicaFetcherThread("test-fetcher", leader, config, failedPartitions, replicaManager, UNBOUNDED_QUOTA, logContext.logPrefix)
+ thread.callShouldFetchFromLastTieredOffset(tp, leaderEndOffset, replicaEndOffset)
+ }
+
+ assertEquals(
+ expected,
+ runScenario(
+ enableLastTieredOffsetFetch = enableLastTieredOffsetFetch,
+ remoteStorageEnabled = remoteStorageEnabled,
+ cleanupPolicy = cleanUpPolicy,
+ leaderEndOffset = leaderEndOffset,
+ replicaEndOffset = replicaEndOffset
+ )
+ )
+ }
+
+ private class MockReplicaFetcherThread(name: String,
Review Comment:
can we remove this `MockReplicaFetcherThread` and expose the `shouldFetchFromLastTieredOffset` in RFT to server package?
```scala
override protected[server] def shouldFetchFromLastTieredOffset(topicPartition: TopicPartition, leaderEndOffset: Long, replicaEndOffset: Long): Boolean
```
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #20428:
URL: https://github.com/apache/kafka/pull/20428#discussion_r2757622885
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -781,7 +787,15 @@ abstract class AbstractFetcherThread(name: String,
leaderEpochInRequest: Optional[Integer],
fetchPartitionData: PartitionData): Boolean = {
try {
- val newFetchState = fetchTierStateMachine.start(topicPartition, fetchState, fetchPartitionData)
+ val fetchFromLastTieredOffset = shouldFetchFromLastTieredOffset(topicPartition, fetchState)
+ val epochAndLogStartOffset = leader.fetchEarliestOffset(topicPartition, fetchState.currentLeaderEpoch())
Review Comment:
`epochAndLogStartOffset` -> `leaderLogStartOffsetAndEpoch`
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -781,7 +787,15 @@ abstract class AbstractFetcherThread(name: String,
leaderEpochInRequest: Optional[Integer],
fetchPartitionData: PartitionData): Boolean = {
try {
- val newFetchState = fetchTierStateMachine.start(topicPartition, fetchState, fetchPartitionData)
+ val fetchFromLastTieredOffset = shouldFetchFromLastTieredOffset(topicPartition, fetchState)
Review Comment:
Shall we rename `fetchFromLastTieredOffset` to `isLastTieredOffsetFetchEnabled`?
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -671,7 +671,13 @@ abstract class AbstractFetcherThread(name: String,
*/
val offsetAndEpoch = leader.fetchLatestOffset(topicPartition, currentLeaderEpoch)
val leaderEndOffset = offsetAndEpoch.offset
- if (leaderEndOffset < replicaEndOffset) {
+ val fetchFromLastTieredOffset = shouldFetchFromLastTieredOffset(topicPartition, leaderEndOffset, replicaEndOffset)
+
+ if (fetchFromLastTieredOffset) {
+ val leaderStartOffset = leader.fetchEarliestOffset(topicPartition, currentLeaderEpoch)
Review Comment:
can `leaderStartOffset` be renamed to `leaderStartOffsetAndEpoch`? And, `epochAndStartingOffset` to `earliestPendingUploadOffsetAndEpoch`
##########
core/src/main/java/kafka/server/TierStateMachine.java:
##########
@@ -182,7 +179,7 @@ private void buildProducerSnapshotFile(UnifiedLog unifiedLog,
* fetching records from the leader. The return value is the next offset to fetch from the leader, which is the
* next offset following the end offset of the remote log portion.
*/
- private Long buildRemoteLogAuxState(TopicPartition topicPartition,
+ protected Long buildRemoteLogAuxState(TopicPartition topicPartition,
Review Comment:
Why do we change the method access specifier from `private` to `protected`?
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -671,7 +671,13 @@ abstract class AbstractFetcherThread(name: String,
*/
val offsetAndEpoch = leader.fetchLatestOffset(topicPartition, currentLeaderEpoch)
val leaderEndOffset = offsetAndEpoch.offset
- if (leaderEndOffset < replicaEndOffset) {
+ val fetchFromLastTieredOffset = shouldFetchFromLastTieredOffset(topicPartition, leaderEndOffset, replicaEndOffset)
+
+ if (fetchFromLastTieredOffset) {
+ val leaderStartOffset = leader.fetchEarliestOffset(topicPartition, currentLeaderEpoch)
+ val epochAndStartingOffset = earliestPendingUploadOffset(topicPartition, currentLeaderEpoch, leaderStartOffset)
+ fetchTierStateMachine.start(topicPartition, topicId.asJava, currentLeaderEpoch, epochAndStartingOffset, leaderStartOffset.offset())
Review Comment:
replace:
topicId.asJava -> topicId.toJava to avoid using deprecated methods
##########
core/src/main/scala/kafka/server/LocalLeaderEndPoint.scala:
##########
@@ -156,7 +156,7 @@ class LocalLeaderEndPoint(sourceBroker: BrokerEndPoint,
new OffsetAndEpoch(-1L, -1)
}
case _ =>
- val earliestPendingUploadOffset = Math.max(highestRemoteOffset + 1, logStartOffset.offset())
+ val earliestPendingUploadOffset = Math.max(highestRemoteOffset + 1, Math.max(logStartOffset.offset(), localLogStartOffset.offset()))
Review Comment:
shall we add an unit test for this case in LocalLeaderEndPointTest?
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -805,6 +819,29 @@ abstract class AbstractFetcherThread(name: String,
}
}
+ /**
+ * Determines the earliest offset for pending uploads, taking into account
+ * both local and remote storage conditions.
+ */
+ private def earliestPendingUploadOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int, leaderLogStartOffset: OffsetAndEpoch): OffsetAndEpoch = {
+ val earliestPendingUploadOffset = leader.fetchEarliestPendingUploadOffset(topicPartition, currentLeaderEpoch)
+ if (earliestPendingUploadOffset.offset == -1L) {
+ val leaderLocalStartOffset = leader.fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch)
+ if (leaderLocalStartOffset.offset == leaderLogStartOffset.offset) {
+ return leaderLocalStartOffset
+ }
+ throw new OffsetNotAvailableException("Segments are uploaded to remote storage, but the leader does not have the information about the uploaded segments")
Review Comment:
Shall we update the error message to be specific?
```
egments are uploaded to remote storage, but the leader does not know the earliest pending upload offset.
```
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -781,7 +787,15 @@ abstract class AbstractFetcherThread(name: String,
leaderEpochInRequest: Optional[Integer],
fetchPartitionData: PartitionData): Boolean = {
try {
- val newFetchState = fetchTierStateMachine.start(topicPartition, fetchState, fetchPartitionData)
+ val fetchFromLastTieredOffset = shouldFetchFromLastTieredOffset(topicPartition, fetchState)
+ val epochAndLogStartOffset = leader.fetchEarliestOffset(topicPartition, fetchState.currentLeaderEpoch())
+ val epochAndStartingOffset = if (fetchFromLastTieredOffset) {
Review Comment:
`epochAndStartingOffset` -> `fetchOffsetAndEpoch`
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -805,6 +819,29 @@ abstract class AbstractFetcherThread(name: String,
}
}
+ /**
+ * Determines the earliest offset for pending uploads, taking into account
+ * both local and remote storage conditions.
+ */
+ private def earliestPendingUploadOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int, leaderLogStartOffset: OffsetAndEpoch): OffsetAndEpoch = {
Review Comment:
can this method be renamed to `fetchEarliestPendingUploadOffset` for clarity? Also, rename the parameter `leaderLogStartOffset` -> `leaderLogStartOffsetAndEpoch`.
##########
core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala:
##########
@@ -18,7 +18,6 @@ package kafka.server
import kafka.cluster.Partition
import kafka.log.LogManager
-
Review Comment:
can this change be avoided?
##########
core/src/main/java/kafka/server/TierStateMachine.java:
##########
@@ -85,21 +85,19 @@ public TierStateMachine(LeaderEndPoint leader,
/**
* Start the tier state machine for the provided topic partition.
*
- * @param topicPartition the topic partition
- * @param currentFetchState the current PartitionFetchState which will
- * be used to derive the return value
- * @param fetchPartitionData the data from the fetch response that returned the offset moved to tiered storage error
- *
- * @return the new PartitionFetchState after the successful start of the
- * tier state machine
+ * @param topicPartition the topic partition for which the tier state machine is to be started
+ * @param topicId the optional unique identifier of the topic
+ * @param currentLeaderEpoch the current leader epoch of the partition
+ * @param epochAndStartingOffset the offset on the leader's local log from which to start replicating logs
+ * @param leaderLogStartOffset the starting offset in the leader's log
+ * @return the new PartitionFetchState after the successful start of the tier state machine
+ * @throws Exception if an error occurs during the process, such as issues with remote storage
*/
PartitionFetchState start(TopicPartition topicPartition,
- PartitionFetchState currentFetchState,
- PartitionData fetchPartitionData) throws Exception {
- OffsetAndEpoch epochAndLeaderLocalStartOffset = leader.fetchEarliestLocalOffset(topicPartition, currentFetchState.currentLeaderEpoch());
- int epoch = epochAndLeaderLocalStartOffset.epoch();
- long leaderLocalStartOffset = epochAndLeaderLocalStartOffset.offset();
-
+ Optional<Uuid> topicId,
+ int currentLeaderEpoch,
+ OffsetAndEpoch epochAndStartingOffset,
Review Comment:
Shall we rename `epochAndStartingOffset` parameter to `fetchOffsetAndEpoch` / `fetchStartOffsetAndEpoch` or something similar?
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1740,4 +1741,1157 @@ class AbstractFetcherThreadTest {
assertEquals(151, replicaState.logEndOffset)
assertEquals(151, replicaState.highWatermark)
}
+
+ @Test
Review Comment:
Please add the Javadoc for the understand the tests.
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #20428:
URL: https://github.com/apache/kafka/pull/20428#discussion_r2757639340
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -805,6 +819,29 @@ abstract class AbstractFetcherThread(name: String,
}
}
+ /**
+ * Determines the earliest offset for pending uploads, taking into account
+ * both local and remote storage conditions.
+ */
+ private def earliestPendingUploadOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int, leaderLogStartOffset: OffsetAndEpoch): OffsetAndEpoch = {
+ val earliestPendingUploadOffset = leader.fetchEarliestPendingUploadOffset(topicPartition, currentLeaderEpoch)
+ if (earliestPendingUploadOffset.offset == -1L) {
+ val leaderLocalStartOffset = leader.fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch)
+ if (leaderLocalStartOffset.offset == leaderLogStartOffset.offset) {
+ return leaderLocalStartOffset
+ }
+ throw new OffsetNotAvailableException("Segments are uploaded to remote storage, but the leader does not have the information about the uploaded segments")
Review Comment:
Shall we update the error message to be specific?
```
Segments are uploaded to remote storage, but the leader does not know the earliest pending upload offset.
```
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on PR #20428:
URL: https://github.com/apache/kafka/pull/20428#issuecomment-3729848465
@HenryCaiHaiying I plan to abandon this PR. I am working on https://github.com/apache/kafka/pull/21154 instead.
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #20428:
URL: https://github.com/apache/kafka/pull/20428#discussion_r2809226577
##########
core/src/main/java/kafka/server/TierStateMachine.java:
##########
@@ -182,7 +179,7 @@ private void buildProducerSnapshotFile(UnifiedLog unifiedLog,
* fetching records from the leader. The return value is the next offset to fetch from the leader, which is the
* next offset following the end offset of the remote log portion.
*/
- private Long buildRemoteLogAuxState(TopicPartition topicPartition,
+ protected Long buildRemoteLogAuxState(TopicPartition topicPartition,
Review Comment:
It is not required any more. Previously we had to override the method in a test. Now that change is not required.
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #20428:
URL: https://github.com/apache/kafka/pull/20428#discussion_r2616755197
##########
core/src/main/scala/kafka/server/LocalLeaderEndPoint.scala:
##########
@@ -135,6 +135,35 @@ class LocalLeaderEndPoint(sourceBroker: BrokerEndPoint,
new OffsetAndEpoch(localLogStartOffset, epoch.orElse(0))
}
+ override def fetchEarliestPendingUploadOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int): OffsetAndEpoch = {
+ val partition = replicaManager.getPartitionOrException(topicPartition)
+ val log = partition.localLogOrException
+
+ if (!log.remoteLogEnabled())
+ return new OffsetAndEpoch(-1L, -1)
+
+ val highestRemoteOffset = log.highestOffsetInRemoteStorage()
+
+ if (highestRemoteOffset == -1L) {
+ val localLogStartOffset = fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch)
+ val logStartOffset = fetchEarliestOffset(topicPartition, currentLeaderEpoch)
+
+ if (localLogStartOffset.offset() == logStartOffset.offset()) {
+ // No segments have been uploaded yet
+ return logStartOffset;
+ } else {
+ // Leader currently does not know about the already uploaded segments
+ return new OffsetAndEpoch(-1L, -1);
+ }
+ }
+
+ val logStartOffset = fetchEarliestOffset(topicPartition, currentLeaderEpoch)
+ val earliestPendingUploadOffset = Math.max(highestRemoteOffset + 1, logStartOffset.offset())
+ val epoch = log.leaderEpochCache.epochForOffset(earliestPendingUploadOffset)
+
+ new OffsetAndEpoch(earliestPendingUploadOffset, epoch.orElse(0))
Review Comment:
I see that for `fetchEarliestLocalOffset` also (line 135), we return 0 when epoch is empty.
--
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-17302: ReplicaFetcher changes for fetching from tiered offset [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #20428:
URL: https://github.com/apache/kafka/pull/20428#issuecomment-3653906803
@abhijeetk88
Could you please split the PR into two to review it properly? Thanks!
PR-1: New methods `LeaderEndPoint#fetchEarliestPendingUploadOffset` and `AbstractFetcherThread#fetchEarliestPendingUploadOffset` with tests.
PR-2: Wiring the actual logic in TierStateMachine and AbstractFetcherThread.
--
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