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 2024/05/24 12:50:21 UTC
[PR] [WIP] KAFKA-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
abhijeetk88 opened a new pull request, #16071:
URL: https://github.com/apache/kafka/pull/16071
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #16071:
URL: https://github.com/apache/kafka/pull/16071#discussion_r1627411601
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1754,12 +1754,27 @@ class ReplicaManager(val config: KafkaConfig,
createLogReadResult(highWatermark, leaderLogStartOffset, leaderLogEndOffset,
new OffsetMovedToTieredStorageException("Given offset" + offset + " is moved to tiered storage"))
} else {
- // For consume fetch requests, create a dummy FetchDataInfo with the remote storage fetch information.
- // For the first topic-partition that needs remote data, we will use this information to read the data in another thread.
- val fetchDataInfo =
- new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY, false, Optional.empty(),
- Optional.of(new RemoteStorageFetchInfo(adjustedMaxBytes, minOneMessage, tp.topicPartition(),
- fetchInfo, params.isolation, params.hardMaxBytesLimit())))
+ val fetchDataInfo = if (remoteLogManager.get.isRemoteLogFetchQuotaExceeded) {
+ // We do not want to send an exception in LogReadResult response (like we do in other cases when we send
+ // UnknownOffsetMetadata), because then it is classified as error in reading data, and a response is
+ // immediately sent back to the client. Instead, we want that we should be able to serve data for the
+ // other topic partitions via delayed fetch if required (when sending immediate response, we skip
+ // delayed fetch). Also, immediately sending response would make the consumer retry again immediately,
+ // which may run into quota exceeded situation again and thus get it into a loop.
+ new FetchDataInfo(
Review Comment:
Note: It makes sense to me to send an empty response for the client to retry fetching the data again from remote storage.
For better clarity, you can reformat the comment like the below or a better statement.
`
We do not want to send an exception in a LogReadResult response (like we do in other cases when we send UnknownOffsetMetadata), because it is classified as an error in reading the data, and a response is immediately sent back to the client. Instead, we want to serve data for the other topic partitions of the fetch request via delayed fetch if required (when sending immediate response, we skip delayed fetch).
`
--
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-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #16071:
URL: https://github.com/apache/kafka/pull/16071#discussion_r1618512042
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -6597,6 +6597,79 @@ class ReplicaManagerTest {
))
}
}
+
+ @Test
+ def testRemoteReadQuotaExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(true)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, fetchInfo.fetchOffsetMetadata)
+ assertFalse(fetchInfo.records.records().iterator().hasNext)
+ assertFalse(fetchInfo.firstEntryIncomplete)
+ assertFalse(fetchInfo.abortedTransactions.isPresent)
+ assertFalse(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ @Test
+ def testRemoteReadQuotaNotExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(false)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(1L, fetchInfo.fetchOffsetMetadata.messageOffset)
+ assertEquals(UnifiedLog.UnknownOffset, fetchInfo.fetchOffsetMetadata.segmentBaseOffset)
+ assertEquals(-1, fetchInfo.fetchOffsetMetadata.relativePositionInSegment)
+ assertEquals(MemoryRecords.EMPTY, fetchInfo.records)
+ assertTrue(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ private def readFromLogWithOffsetOutOfRange(tp: TopicPartition): Seq[(TopicIdPartition, LogReadResult)] = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), aliveBrokerIds = Seq(0, 1, 2), enableRemoteStorage = true, shouldMockLog = true)
+ try {
+ val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)
+ replicaManager.createPartition(tp).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints = offsetCheckpoints, None)
+ val partition0Replicas = Seq[Integer](0, 1).asJava
+ val topicIds = Map(tp.topic -> topicId).asJava
+ val leaderEpoch = 0
+ val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
+ Seq(
+ new LeaderAndIsrPartitionState()
+ .setTopicName(tp.topic)
+ .setPartitionIndex(tp.partition)
+ .setControllerEpoch(0)
+ .setLeader(leaderEpoch)
+ .setLeaderEpoch(0)
+ .setIsr(partition0Replicas)
+ .setPartitionEpoch(0)
+ .setReplicas(partition0Replicas)
+ .setIsNew(true)
+ ).asJava,
+ topicIds,
+ Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
+ replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest, (_, _) => ())
+
+ val params = new FetchParams(ApiKeys.FETCH.latestVersion, -1, 1, 1000, 0, 100, FetchIsolation.HIGH_WATERMARK, None.asJava)
+ replicaManager.readFromLog(
+ params,
+ Seq(new TopicIdPartition(topicId, 0, topic) -> new PartitionData(topicId, 1, 0, 100000, Optional.of[Integer](leaderEpoch), Optional.of(leaderEpoch))),
Review Comment:
The log is not actually empty. The log start offset is 0 here. Since the fetch offset is 1, data from remote storage needs to be fetched.
--
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-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #16071:
URL: https://github.com/apache/kafka/pull/16071#issuecomment-2149976460
There were a couple of unrelated test failures, merging it to trunk and 3.8 branches.
--
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-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #16071:
URL: https://github.com/apache/kafka/pull/16071#discussion_r1618482187
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -6597,6 +6597,79 @@ class ReplicaManagerTest {
))
}
}
+
+ @Test
+ def testRemoteReadQuotaExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(true)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, fetchInfo.fetchOffsetMetadata)
+ assertFalse(fetchInfo.records.records().iterator().hasNext)
+ assertFalse(fetchInfo.firstEntryIncomplete)
+ assertFalse(fetchInfo.abortedTransactions.isPresent)
+ assertFalse(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ @Test
+ def testRemoteReadQuotaNotExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(false)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(1L, fetchInfo.fetchOffsetMetadata.messageOffset)
+ assertEquals(UnifiedLog.UnknownOffset, fetchInfo.fetchOffsetMetadata.segmentBaseOffset)
+ assertEquals(-1, fetchInfo.fetchOffsetMetadata.relativePositionInSegment)
+ assertEquals(MemoryRecords.EMPTY, fetchInfo.records)
+ assertTrue(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ private def readFromLogWithOffsetOutOfRange(tp: TopicPartition): Seq[(TopicIdPartition, LogReadResult)] = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), aliveBrokerIds = Seq(0, 1, 2), enableRemoteStorage = true, shouldMockLog = true)
+ try {
+ val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)
+ replicaManager.createPartition(tp).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints = offsetCheckpoints, None)
+ val partition0Replicas = Seq[Integer](0, 1).asJava
+ val topicIds = Map(tp.topic -> topicId).asJava
+ val leaderEpoch = 0
+ val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
+ Seq(
+ new LeaderAndIsrPartitionState()
+ .setTopicName(tp.topic)
+ .setPartitionIndex(tp.partition)
+ .setControllerEpoch(0)
+ .setLeader(leaderEpoch)
Review Comment:
Thanks for pointing this out. I missed this when I copied this snippet from other 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-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #16071:
URL: https://github.com/apache/kafka/pull/16071#discussion_r1617245864
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -6597,6 +6597,79 @@ class ReplicaManagerTest {
))
}
}
+
+ @Test
+ def testRemoteReadQuotaExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(true)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, fetchInfo.fetchOffsetMetadata)
+ assertFalse(fetchInfo.records.records().iterator().hasNext)
+ assertFalse(fetchInfo.firstEntryIncomplete)
+ assertFalse(fetchInfo.abortedTransactions.isPresent)
+ assertFalse(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ @Test
+ def testRemoteReadQuotaNotExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(false)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(1L, fetchInfo.fetchOffsetMetadata.messageOffset)
+ assertEquals(UnifiedLog.UnknownOffset, fetchInfo.fetchOffsetMetadata.segmentBaseOffset)
+ assertEquals(-1, fetchInfo.fetchOffsetMetadata.relativePositionInSegment)
+ assertEquals(MemoryRecords.EMPTY, fetchInfo.records)
+ assertTrue(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ private def readFromLogWithOffsetOutOfRange(tp: TopicPartition): Seq[(TopicIdPartition, LogReadResult)] = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), aliveBrokerIds = Seq(0, 1, 2), enableRemoteStorage = true, shouldMockLog = true)
+ try {
+ val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)
+ replicaManager.createPartition(tp).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints = offsetCheckpoints, None)
+ val partition0Replicas = Seq[Integer](0, 1).asJava
+ val topicIds = Map(tp.topic -> topicId).asJava
+ val leaderEpoch = 0
+ val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
+ Seq(
+ new LeaderAndIsrPartitionState()
+ .setTopicName(tp.topic)
+ .setPartitionIndex(tp.partition)
+ .setControllerEpoch(0)
+ .setLeader(leaderEpoch)
Review Comment:
can we hardcode (or) define a new variable for `leader` instead of reusing the `leaderEpoch`?
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -6597,6 +6597,79 @@ class ReplicaManagerTest {
))
}
}
+
+ @Test
+ def testRemoteReadQuotaExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(true)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, fetchInfo.fetchOffsetMetadata)
+ assertFalse(fetchInfo.records.records().iterator().hasNext)
+ assertFalse(fetchInfo.firstEntryIncomplete)
+ assertFalse(fetchInfo.abortedTransactions.isPresent)
+ assertFalse(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ @Test
+ def testRemoteReadQuotaNotExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(false)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(1L, fetchInfo.fetchOffsetMetadata.messageOffset)
+ assertEquals(UnifiedLog.UnknownOffset, fetchInfo.fetchOffsetMetadata.segmentBaseOffset)
+ assertEquals(-1, fetchInfo.fetchOffsetMetadata.relativePositionInSegment)
+ assertEquals(MemoryRecords.EMPTY, fetchInfo.records)
+ assertTrue(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ private def readFromLogWithOffsetOutOfRange(tp: TopicPartition): Seq[(TopicIdPartition, LogReadResult)] = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), aliveBrokerIds = Seq(0, 1, 2), enableRemoteStorage = true, shouldMockLog = true)
+ try {
+ val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)
+ replicaManager.createPartition(tp).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints = offsetCheckpoints, None)
+ val partition0Replicas = Seq[Integer](0, 1).asJava
+ val topicIds = Map(tp.topic -> topicId).asJava
+ val leaderEpoch = 0
+ val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
+ Seq(
+ new LeaderAndIsrPartitionState()
+ .setTopicName(tp.topic)
+ .setPartitionIndex(tp.partition)
+ .setControllerEpoch(0)
+ .setLeader(leaderEpoch)
+ .setLeaderEpoch(0)
+ .setIsr(partition0Replicas)
+ .setPartitionEpoch(0)
+ .setReplicas(partition0Replicas)
+ .setIsNew(true)
+ ).asJava,
+ topicIds,
+ Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
+ replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest, (_, _) => ())
+
+ val params = new FetchParams(ApiKeys.FETCH.latestVersion, -1, 1, 1000, 0, 100, FetchIsolation.HIGH_WATERMARK, None.asJava)
+ replicaManager.readFromLog(
+ params,
+ Seq(new TopicIdPartition(topicId, 0, topic) -> new PartitionData(topicId, 1, 0, 100000, Optional.of[Integer](leaderEpoch), Optional.of(leaderEpoch))),
Review Comment:
The test exposes an issue, the `fetchOffset` is 1 (since the log is empty) and the replica-manager throws OffsetOutOfRangeException. Why we do have to check the remoteLogFetch quota in this case?
--
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-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd merged PR #16071:
URL: https://github.com/apache/kafka/pull/16071
--
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-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #16071:
URL: https://github.com/apache/kafka/pull/16071#discussion_r1627421355
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1754,12 +1754,27 @@ class ReplicaManager(val config: KafkaConfig,
createLogReadResult(highWatermark, leaderLogStartOffset, leaderLogEndOffset,
new OffsetMovedToTieredStorageException("Given offset" + offset + " is moved to tiered storage"))
} else {
- // For consume fetch requests, create a dummy FetchDataInfo with the remote storage fetch information.
- // For the first topic-partition that needs remote data, we will use this information to read the data in another thread.
- val fetchDataInfo =
- new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY, false, Optional.empty(),
- Optional.of(new RemoteStorageFetchInfo(adjustedMaxBytes, minOneMessage, tp.topicPartition(),
- fetchInfo, params.isolation, params.hardMaxBytesLimit())))
+ val fetchDataInfo = if (remoteLogManager.get.isRemoteLogFetchQuotaExceeded) {
+ // We do not want to send an exception in LogReadResult response (like we do in other cases when we send
+ // UnknownOffsetMetadata), because then it is classified as error in reading data, and a response is
+ // immediately sent back to the client. Instead, we want that we should be able to serve data for the
+ // other topic partitions via delayed fetch if required (when sending immediate response, we skip
+ // delayed fetch). Also, immediately sending response would make the consumer retry again immediately,
+ // which may run into quota exceeded situation again and thus get it into a loop.
+ new FetchDataInfo(
Review Comment:
Done.
--
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-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #16071:
URL: https://github.com/apache/kafka/pull/16071#discussion_r1621675519
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -6597,6 +6597,79 @@ class ReplicaManagerTest {
))
}
}
+
+ @Test
+ def testRemoteReadQuotaExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(true)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, fetchInfo.fetchOffsetMetadata)
+ assertFalse(fetchInfo.records.records().iterator().hasNext)
+ assertFalse(fetchInfo.firstEntryIncomplete)
+ assertFalse(fetchInfo.abortedTransactions.isPresent)
+ assertFalse(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ @Test
+ def testRemoteReadQuotaNotExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(false)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(1L, fetchInfo.fetchOffsetMetadata.messageOffset)
+ assertEquals(UnifiedLog.UnknownOffset, fetchInfo.fetchOffsetMetadata.segmentBaseOffset)
+ assertEquals(-1, fetchInfo.fetchOffsetMetadata.relativePositionInSegment)
+ assertEquals(MemoryRecords.EMPTY, fetchInfo.records)
+ assertTrue(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ private def readFromLogWithOffsetOutOfRange(tp: TopicPartition): Seq[(TopicIdPartition, LogReadResult)] = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), aliveBrokerIds = Seq(0, 1, 2), enableRemoteStorage = true, shouldMockLog = true)
+ try {
+ val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)
+ replicaManager.createPartition(tp).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints = offsetCheckpoints, None)
+ val partition0Replicas = Seq[Integer](0, 1).asJava
+ val topicIds = Map(tp.topic -> topicId).asJava
+ val leaderEpoch = 0
+ val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
+ Seq(
+ new LeaderAndIsrPartitionState()
+ .setTopicName(tp.topic)
+ .setPartitionIndex(tp.partition)
+ .setControllerEpoch(0)
+ .setLeader(leaderEpoch)
Review Comment:
done
--
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-15265: Integrate RLMQuotaManager for throttling fetches from remote storage [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #16071:
URL: https://github.com/apache/kafka/pull/16071#discussion_r1617253547
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -6597,6 +6597,79 @@ class ReplicaManagerTest {
))
}
}
+
+ @Test
+ def testRemoteReadQuotaExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(true)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, fetchInfo.fetchOffsetMetadata)
+ assertFalse(fetchInfo.records.records().iterator().hasNext)
+ assertFalse(fetchInfo.firstEntryIncomplete)
+ assertFalse(fetchInfo.abortedTransactions.isPresent)
+ assertFalse(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ @Test
+ def testRemoteReadQuotaNotExceeded(): Unit = {
+ when(mockRemoteLogManager.isRemoteLogFetchQuotaExceeded).thenReturn(false)
+
+ val tp0 = new TopicPartition(topic, 0)
+ val tpId0 = new TopicIdPartition(topicId, tp0)
+ val fetch: Seq[(TopicIdPartition, LogReadResult)] = readFromLogWithOffsetOutOfRange(tp0)
+
+ assertEquals(1, fetch.size)
+ assertEquals(tpId0, fetch.head._1)
+ val fetchInfo = fetch.head._2.info
+ assertEquals(1L, fetchInfo.fetchOffsetMetadata.messageOffset)
+ assertEquals(UnifiedLog.UnknownOffset, fetchInfo.fetchOffsetMetadata.segmentBaseOffset)
+ assertEquals(-1, fetchInfo.fetchOffsetMetadata.relativePositionInSegment)
+ assertEquals(MemoryRecords.EMPTY, fetchInfo.records)
+ assertTrue(fetchInfo.delayedRemoteStorageFetch.isPresent)
+ }
+
+ private def readFromLogWithOffsetOutOfRange(tp: TopicPartition): Seq[(TopicIdPartition, LogReadResult)] = {
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), aliveBrokerIds = Seq(0, 1, 2), enableRemoteStorage = true, shouldMockLog = true)
+ try {
+ val offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)
+ replicaManager.createPartition(tp).createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints = offsetCheckpoints, None)
+ val partition0Replicas = Seq[Integer](0, 1).asJava
+ val topicIds = Map(tp.topic -> topicId).asJava
+ val leaderEpoch = 0
+ val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
+ Seq(
+ new LeaderAndIsrPartitionState()
+ .setTopicName(tp.topic)
+ .setPartitionIndex(tp.partition)
+ .setControllerEpoch(0)
+ .setLeader(leaderEpoch)
+ .setLeaderEpoch(0)
+ .setIsr(partition0Replicas)
+ .setPartitionEpoch(0)
+ .setReplicas(partition0Replicas)
+ .setIsNew(true)
+ ).asJava,
+ topicIds,
+ Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
+ replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest, (_, _) => ())
+
+ val params = new FetchParams(ApiKeys.FETCH.latestVersion, -1, 1, 1000, 0, 100, FetchIsolation.HIGH_WATERMARK, None.asJava)
+ replicaManager.readFromLog(
+ params,
+ Seq(new TopicIdPartition(topicId, 0, topic) -> new PartitionData(topicId, 1, 0, 100000, Optional.of[Integer](leaderEpoch), Optional.of(leaderEpoch))),
Review Comment:
The test exposes an issue, the `fetchOffset` is 1 and the replica-manager throws OffsetOutOfRangeException (since the log is empty) . Why we do have to check the remoteLogFetch quota in this case?
--
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