You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/06/02 09:47:13 UTC
(kafka) branch 3.7 updated: KAFKA-16807 DescribeLogDirsResponseData#results#topics have unexpected topics having empty partitions (#16042)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 3.7
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.7 by this push:
new 1e9d0c4cd25 KAFKA-16807 DescribeLogDirsResponseData#results#topics have unexpected topics having empty partitions (#16042)
1e9d0c4cd25 is described below
commit 1e9d0c4cd25876c44b2e579d9d5e99746cbf660b
Author: Ken Huang <10...@users.noreply.github.com>
AuthorDate: Sun Jun 2 18:33:02 2024 +0900
KAFKA-16807 DescribeLogDirsResponseData#results#topics have unexpected topics having empty partitions (#16042)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../main/scala/kafka/server/ReplicaManager.scala | 25 ++++++++--------
.../unit/kafka/server/ReplicaManagerTest.scala | 33 ++++++++++++++++++++++
2 files changed, 47 insertions(+), 11 deletions(-)
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index fa0eb0d8581..d46ca67e538 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -33,6 +33,7 @@ import kafka.zk.KafkaZkClient
import org.apache.kafka.common.errors._
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.DeleteRecordsResponseData.DeleteRecordsPartitionResult
+import org.apache.kafka.common.message.DescribeLogDirsResponseData.DescribeLogDirsTopic
import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState
import org.apache.kafka.common.message.LeaderAndIsrResponseData.{LeaderAndIsrPartitionError, LeaderAndIsrTopicError}
import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetForLeaderTopic
@@ -67,7 +68,7 @@ import java.util
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.locks.Lock
import java.util.concurrent.{CompletableFuture, Future, RejectedExecutionException, TimeUnit}
-import java.util.{Optional, OptionalInt, OptionalLong}
+import java.util.{Collections, Optional, OptionalInt, OptionalLong}
import scala.collection.{Map, Seq, Set, mutable}
import scala.compat.java8.OptionConverters._
import scala.jdk.CollectionConverters._
@@ -1383,9 +1384,9 @@ class ReplicaManager(val config: KafkaConfig,
val fileStore = Files.getFileStore(file)
val totalBytes = adjustForLargeFileSystems(fileStore.getTotalSpace)
val usableBytes = adjustForLargeFileSystems(fileStore.getUsableSpace)
- logsByDir.get(absolutePath) match {
+ val topicInfos = logsByDir.get(absolutePath) match {
case Some(logs) =>
- val topicInfos = logs.groupBy(_.topicPartition.topic).map{case (topic, logs) =>
+ logs.groupBy(_.topicPartition.topic).map { case (topic, logs) =>
new DescribeLogDirsResponseData.DescribeLogDirsTopic().setName(topic).setPartitions(
logs.filter { log =>
partitions.contains(log.topicPartition)
@@ -1396,17 +1397,19 @@ class ReplicaManager(val config: KafkaConfig,
.setOffsetLag(getLogEndOffsetLag(log.topicPartition, log.logEndOffset, log.isFuture))
.setIsFutureKey(log.isFuture)
}.toList.asJava)
- }.toList.asJava
-
- new DescribeLogDirsResponseData.DescribeLogDirsResult().setLogDir(absolutePath)
- .setErrorCode(Errors.NONE.code).setTopics(topicInfos)
- .setTotalBytes(totalBytes).setUsableBytes(usableBytes)
+ }.filterNot(_.partitions().isEmpty).toList.asJava
case None =>
- new DescribeLogDirsResponseData.DescribeLogDirsResult().setLogDir(absolutePath)
- .setErrorCode(Errors.NONE.code)
- .setTotalBytes(totalBytes).setUsableBytes(usableBytes)
+ Collections.emptyList[DescribeLogDirsTopic]()
}
+ val describeLogDirsResult = new DescribeLogDirsResponseData.DescribeLogDirsResult()
+ .setLogDir(absolutePath).setTopics(topicInfos)
+ .setErrorCode(Errors.NONE.code)
+ .setTotalBytes(totalBytes).setUsableBytes(usableBytes)
+ if (!topicInfos.isEmpty)
+ describeLogDirsResult.setTopics(topicInfos)
+ describeLogDirsResult
+
} catch {
case e: KafkaStorageException =>
warn("Unable to describe replica dirs for %s".format(absolutePath), e)
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index c13043f2d74..f3336a70fed 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -6378,6 +6378,39 @@ class ReplicaManagerTest {
assertEquals(Errors.NONE.code, response.errorCode)
assertTrue(response.totalBytes > 0)
assertTrue(response.usableBytes >= 0)
+ assertFalse(response.topics().isEmpty)
+ response.topics().forEach(t => assertFalse(t.partitions().isEmpty))
+ }
+ } finally {
+ replicaManager.shutdown(checkpointHW = false)
+ }
+ }
+
+ @Test
+ def testDescribeLogDirsWithoutAnyPartitionTopic(): Unit = {
+ val noneTopic = "none-topic"
+ val topicPartition = 0
+ val topicId = Uuid.randomUuid()
+ val followerBrokerId = 0
+ val leaderBrokerId = 1
+ val leaderEpoch = 1
+ val leaderEpochIncrement = 2
+ val countDownLatch = new CountDownLatch(1)
+ val offsetFromLeader = 5
+
+ // Prepare the mocked components for the test
+ val (replicaManager, mockLogMgr) = prepareReplicaManagerAndLogManager(new MockTimer(time),
+ topicPartition, leaderEpoch + leaderEpochIncrement, followerBrokerId, leaderBrokerId, countDownLatch,
+ expectTruncation = false, localLogOffset = Some(10), offsetFromLeader = offsetFromLeader, topicId = Some(topicId))
+
+ try {
+ val responses = replicaManager.describeLogDirs(Set(new TopicPartition(noneTopic, topicPartition)))
+ assertEquals(mockLogMgr.liveLogDirs.size, responses.size)
+ responses.foreach { response =>
+ assertEquals(Errors.NONE.code, response.errorCode)
+ assertTrue(response.totalBytes > 0)
+ assertTrue(response.usableBytes >= 0)
+ assertTrue(response.topics().isEmpty)
}
} finally {
replicaManager.shutdown(checkpointHW = false)