KAFKA-18555: Avoid casting MetadataCache to KRaftMetadataCache #18632

FrankYang0529 · 2025-01-20T02:22:30Z

The ZkMetadataCache is removed in https://issues.apache.org/jira/browse/KAFKA-18373. We should pull up the relevant methods so that no casting from MetadataCache to KRaftMetadataCache is required.

Add following methods to MetadataCache:

getAliveBrokerEpoch
isBrokerFenced
isBrokerShuttingDown
describeTopicResponse

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

Signed-off-by: PoAn Yang <[email protected]>

FrankYang0529 · 2025-01-20T02:35:03Z

@ijuma The only remaining casting is in DescribeTopicPartitionsRequestHandler. It uses KRaftMetadataCache#getTopicMetadataForDescribeTopicResponse. Do you think that we should either add it to MetadataCache or refactoring it to move related logic to DescribeTopicPartitionsRequestHandler? Or we can just keep it as current implementation? Thanks.

  val describeTopicPartitionsRequestHandler : DescribeTopicPartitionsRequestHandler = new DescribeTopicPartitionsRequestHandler(
    metadataCache.asInstanceOf[KRaftMetadataCache], authHelper, config)

ijuma · 2025-01-20T09:29:29Z

It's ok to pull up that method to MetadataCache, but we should take the chance and make it more concise. Instead of metadataCache.getTopicMetadataForDescribeTopicResponse, we can simply call it metadataCache.describeTopicResponse.

Signed-off-by: PoAn Yang <[email protected]>

ijuma

Thanks, left a few comments.

ijuma · 2025-01-22T14:32:06Z

core/src/main/scala/kafka/server/KafkaApis.scala

-      Some(new DescribeTopicPartitionsRequestHandler(kRaftMetadataCache, authHelper, config))
-    case _ => None
-  }
+  val describeTopicPartitionsRequestHandler : DescribeTopicPartitionsRequestHandler = new DescribeTopicPartitionsRequestHandler(


This is an existing issue, but since you're touching this file, remove the : DescribeTopicPartitionsRequestHandler (it's redundant).

ijuma · 2025-01-22T14:33:21Z

core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala

@@ -272,7 +272,7 @@ class KRaftMetadataCache(
   * @param maximumNumberOfPartitions     The max number of partitions to return.
   * @param ignoreTopicsWithExceptions    Whether ignore the topics with exception.
   */


Move this scaladoc to the interface.

@FrankYang0529 Could you please remove those comments as you have moved them to the interface

ijuma · 2025-01-22T14:33:59Z

core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java

-        KRaftMetadataCache metadataCache = new KRaftMetadataCache(0, () -> KRaftVersion.KRAFT_VERSION_1);
-        updateKraftMetadataCache(metadataCache, records);
+        MetadataCache metadataCache = new KRaftMetadataCache(0, () -> KRaftVersion.KRAFT_VERSION_1);
+        updateKraftMetadataCache((KRaftMetadataCache) metadataCache, records);


Both of these changes seem to make things worse.

ijuma · 2025-01-22T14:34:07Z

core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java

-        KRaftMetadataCache metadataCache = new KRaftMetadataCache(0, () -> KRaftVersion.KRAFT_VERSION_1);
-        updateKraftMetadataCache(metadataCache, records);
+        MetadataCache metadataCache = new KRaftMetadataCache(0, () -> KRaftVersion.KRAFT_VERSION_1);
+        updateKraftMetadataCache((KRaftMetadataCache) metadataCache, records);


Same as the comment above.

ijuma · 2025-01-22T14:34:25Z

core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala

@@ -348,6 +348,7 @@ class PartitionLockTest extends Logging {
    val controllerEpoch = 0
    val replicas = (0 to numReplicaFetchers).map(i => Integer.valueOf(brokerId + i)).toList.asJava
    val isr = replicas
+    replicas.forEach(replicaId => when(metadataCache.getAliveBrokerEpoch(replicaId)).thenReturn(Some(1L)))


Why do we need this extra code?

We use mock(classOf[MetadataCache]) for metadataCache, so it didn't really get into KRaftMetadataCache block. In this PR, we remove match block, so we have to add related mock.

kafka/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala

Line 270 in 94a1bfb

val metadataCache: MetadataCache = mock(classOf[MetadataCache])

kafka/core/src/main/scala/kafka/cluster/Replica.scala

Lines 116 to 125 in 94a1bfb

metadataCache match {

case kRaftMetadataCache: KRaftMetadataCache =>

val cachedBrokerEpoch = kRaftMetadataCache.getAliveBrokerEpoch(brokerId)

// Fence the update if it provides a stale broker epoch.

if (brokerEpoch != -1 && cachedBrokerEpoch.exists(_ > brokerEpoch)) {

throw new NotLeaderOrFollowerException(s"Received stale fetch state update. broker epoch=$brokerEpoch " +

s"vs expected=${cachedBrokerEpoch.get}")

}

case _ =>

}

ijuma · 2025-01-22T14:35:12Z

core/src/test/scala/unit/kafka/cluster/PartitionTest.scala

@@ -186,6 +186,7 @@ class PartitionTest extends AbstractPartitionTest {
    val leaderEpoch = 10
    val logStartOffset = 0L
    val partition = setupPartitionWithMocks(leaderEpoch = leaderEpoch, isLeader = true)
+    addBrokerEpochToMockMetadataCache(metadataCache, List(remoteReplicaId))


Hmm, why are these changes needed?

Same as above. We use mock(classOf[MetadataCache]) in AbstractPartitionTest.

kafka/core/src/test/scala/unit/kafka/cluster/AbstractPartitionTest.scala

Line 64 in 94a1bfb

val metadataCache: MetadataCache = mock(classOf[MetadataCache])

ijuma · 2025-01-22T14:35:55Z

core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala

@@ -793,6 +793,7 @@ class ReplicaManagerTest {

    try {
      val brokerList = Seq[Integer](0, 1).asJava
+      when(replicaManager.metadataCache.getAliveBrokerEpoch(1)).thenReturn(Some(brokerEpoch))


Why is this change needed?

Same as above. In ReplicaManagerTest#setupReplicaManagerWithMockedPurgatories, it uses mock(classOf[MetadataCache]).

kafka/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala

Line 3141 in 94a1bfb

val metadataCache: MetadataCache = mock(classOf[MetadataCache])

Could you please move the mock to line#3154? We should mock the method after creating the mocked metadataCache

Signed-off-by: PoAn Yang <[email protected]>

chia7712 · 2025-01-23T11:40:09Z

core/src/main/scala/kafka/cluster/Partition.scala

-          isBrokerEpochIsrEligible(storedBrokerEpoch, cachedBrokerEpoch)
-
-      case _ => true
+    // In KRaft mode, only a replica which meets all of the following requirements is allowed to join the ISR.


Maybe we can remove "In KRaft mode"?

Signed-off-by: PoAn Yang <[email protected]>

chia7712

@FrankYang0529 could you please fix https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/cluster/Partition.scala#L1814?

Signed-off-by: PoAn Yang <[email protected]>

FrankYang0529 · 2025-01-24T06:04:24Z

could you please fix https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/cluster/Partition.scala#L1814?

Updated it. Thanks!

chia7712

@FrankYang0529 overall LGTM except for one trivial comment

chia7712 · 2025-01-24T10:44:06Z

core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala

@@ -272,7 +272,7 @@ class KRaftMetadataCache(
   * @param maximumNumberOfPartitions     The max number of partitions to return.
   * @param ignoreTopicsWithExceptions    Whether ignore the topics with exception.
   */


@FrankYang0529 Could you please remove those comments as you have moved them to the interface

Signed-off-by: PoAn Yang <[email protected]>

KAFKA-18555: Avoid casting MetadataCache to KRaftMetadataCache

0f3d5c5

Signed-off-by: PoAn Yang <[email protected]>

github-actions bot added the core Kafka Broker label Jan 20, 2025

FrankYang0529 added 5 commits January 20, 2025 22:30

fix CI error and address comment

333bbc1

Signed-off-by: PoAn Yang <[email protected]>

Merge branch 'trunk' into KAFKA-18555

32e0d48

fix CI error

5858144

Signed-off-by: PoAn Yang <[email protected]>

Merge branch 'trunk' into KAFKA-18555

dd31568

fix CI error

5b3f501

Signed-off-by: PoAn Yang <[email protected]>

FrankYang0529 requested a review from ijuma January 21, 2025 03:27

ijuma reviewed Jan 22, 2025

View reviewed changes

FrankYang0529 added 2 commits January 23, 2025 10:10

Merge branch 'trunk' into KAFKA-18555

69b9b19

address comment

755c781

Signed-off-by: PoAn Yang <[email protected]>

chia7712 reviewed Jan 23, 2025

View reviewed changes

FrankYang0529 added 2 commits January 23, 2025 21:57

Merge branch 'trunk' into KAFKA-18555

c8c1ccd

address comment

67936d7

Signed-off-by: PoAn Yang <[email protected]>

chia7712 reviewed Jan 23, 2025

View reviewed changes

chia7712 mentioned this pull request Jan 23, 2025

KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft #18196

Open

3 tasks

FrankYang0529 added 2 commits January 24, 2025 09:14

address comment

ac6c9b2

Signed-off-by: PoAn Yang <[email protected]>

Merge branch 'trunk' into KAFKA-18555

4206047

chia7712 reviewed Jan 24, 2025

View reviewed changes

FrankYang0529 added 2 commits January 24, 2025 21:01

Merge branch 'trunk' into KAFKA-18555

2f42b87

address comment

8ea7746

Signed-off-by: PoAn Yang <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18555: Avoid casting MetadataCache to KRaftMetadataCache #18632

KAFKA-18555: Avoid casting MetadataCache to KRaftMetadataCache #18632

FrankYang0529 commented Jan 20, 2025 •

edited

Loading

FrankYang0529 commented Jan 20, 2025

ijuma commented Jan 20, 2025 •

edited

Loading

ijuma left a comment

ijuma Jan 22, 2025

ijuma Jan 22, 2025

chia7712 Jan 24, 2025

ijuma Jan 22, 2025

ijuma Jan 22, 2025

ijuma Jan 22, 2025

FrankYang0529 Jan 23, 2025

ijuma Jan 22, 2025

FrankYang0529 Jan 23, 2025

ijuma Jan 22, 2025

FrankYang0529 Jan 23, 2025

chia7712 Jan 23, 2025

chia7712 Jan 23, 2025

chia7712 left a comment

FrankYang0529 commented Jan 24, 2025

chia7712 left a comment

chia7712 Jan 24, 2025

	metadataCache match {
	case kRaftMetadataCache: KRaftMetadataCache =>
	val cachedBrokerEpoch = kRaftMetadataCache.getAliveBrokerEpoch(brokerId)
	// Fence the update if it provides a stale broker epoch.
	if (brokerEpoch != -1 && cachedBrokerEpoch.exists(_ > brokerEpoch)) {
	throw new NotLeaderOrFollowerException(s"Received stale fetch state update. broker epoch=$brokerEpoch " +
	s"vs expected=${cachedBrokerEpoch.get}")
	}
	case _ =>
	}

KAFKA-18555: Avoid casting MetadataCache to KRaftMetadataCache #18632

Are you sure you want to change the base?

KAFKA-18555: Avoid casting MetadataCache to KRaftMetadataCache #18632

Conversation

FrankYang0529 commented Jan 20, 2025 • edited Loading

Committer Checklist (excluded from commit message)

FrankYang0529 commented Jan 20, 2025

ijuma commented Jan 20, 2025 • edited Loading

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

FrankYang0529 commented Jan 24, 2025

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

FrankYang0529 commented Jan 20, 2025 •

edited

Loading

ijuma commented Jan 20, 2025 •

edited

Loading