You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2024/03/30 11:18:47 UTC
(kafka) branch trunk updated: KAFKA-16447: Fix failed ReplicaManagerTest (#15630)
This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 9a9b532d5d5 KAFKA-16447: Fix failed ReplicaManagerTest (#15630)
9a9b532d5d5 is described below
commit 9a9b532d5d5beeecf7a4b769731ee609625429e1
Author: PoAn Yang <pa...@apache.org>
AuthorDate: Sat Mar 30 19:18:41 2024 +0800
KAFKA-16447: Fix failed ReplicaManagerTest (#15630)
The change in https://github.com/apache/kafka/pull/15373/files#r1544335647 updated broker port from 9093 to 9094. Some of test cases check broker endpoint with fixed string 9093. I change test cases to get endpoint by broker id, so these cases will not fail if someone change the port again in the future.
Reviewers: Luke Chen <sh...@gmail.com>
---
.../unit/kafka/server/ReplicaManagerTest.scala | 28 ++++++++++++++--------
1 file changed, 18 insertions(+), 10 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index ed64d986c26..35a5d9cc7fb 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -5161,7 +5161,8 @@ class ReplicaManagerTest {
}
val fetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
- assertEquals(Some(BrokerEndPoint(otherId, "localhost", 9093)), fetcher.map(_.leader.brokerEndPoint()))
+ val otherEndpoint = ClusterImageTest.IMAGE1.broker(otherId).listeners().get("PLAINTEXT")
+ assertEquals(Some(BrokerEndPoint(otherId, otherEndpoint.host(), otherEndpoint.port())), fetcher.map(_.leader.brokerEndPoint()))
} finally {
replicaManager.shutdown(checkpointHW = false)
}
@@ -5193,7 +5194,8 @@ class ReplicaManagerTest {
}
val fetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
- assertEquals(Some(BrokerEndPoint(otherId, "localhost", 9093)), fetcher.map(_.leader.brokerEndPoint()))
+ val otherEndpoint = ClusterImageTest.IMAGE1.broker(otherId).listeners().get("PLAINTEXT")
+ assertEquals(Some(BrokerEndPoint(otherId, otherEndpoint.host(), otherEndpoint.port())), fetcher.map(_.leader.brokerEndPoint()))
// Append on a follower should fail
val followerResponse = sendProducerAppend(replicaManager, topicPartition, numOfRecords)
@@ -5255,7 +5257,8 @@ class ReplicaManagerTest {
}
val fetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
- assertEquals(Some(BrokerEndPoint(otherId, "localhost", 9093)), fetcher.map(_.leader.brokerEndPoint()))
+ val otherEndpoint = ClusterImageTest.IMAGE1.broker(otherId).listeners().get("PLAINTEXT")
+ assertEquals(Some(BrokerEndPoint(otherId, otherEndpoint.host(), otherEndpoint.port())), fetcher.map(_.leader.brokerEndPoint()))
// Apply the same delta again
replicaManager.applyDelta(followerTopicsDelta, followerMetadataImage)
@@ -5270,7 +5273,7 @@ class ReplicaManagerTest {
}
val noChangeFetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
- assertEquals(Some(BrokerEndPoint(otherId, "localhost", 9093)), noChangeFetcher.map(_.leader.brokerEndPoint()))
+ assertEquals(Some(BrokerEndPoint(otherId, otherEndpoint.host(), otherEndpoint.port())), noChangeFetcher.map(_.leader.brokerEndPoint()))
} finally {
replicaManager.shutdown(checkpointHW = false)
}
@@ -5301,7 +5304,8 @@ class ReplicaManagerTest {
}
val fetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
- assertEquals(Some(BrokerEndPoint(otherId, "localhost", 9093)), fetcher.map(_.leader.brokerEndPoint()))
+ val otherEndpoint = ClusterImageTest.IMAGE1.broker(otherId).listeners().get("PLAINTEXT")
+ assertEquals(Some(BrokerEndPoint(otherId, otherEndpoint.host(), otherEndpoint.port())), fetcher.map(_.leader.brokerEndPoint()))
// Apply changes that remove replica
val notReplicaTopicsDelta = topicsChangeDelta(followerMetadataImage.topics(), otherId, true)
@@ -5348,7 +5352,8 @@ class ReplicaManagerTest {
}
val fetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
- assertEquals(Some(BrokerEndPoint(otherId, "localhost", 9093)), fetcher.map(_.leader.brokerEndPoint()))
+ val otherEndpoint = ClusterImageTest.IMAGE1.broker(otherId).listeners().get("PLAINTEXT")
+ assertEquals(Some(BrokerEndPoint(otherId, otherEndpoint.host(), otherEndpoint.port())), fetcher.map(_.leader.brokerEndPoint()))
// Apply changes that remove topic and replica
val removeTopicsDelta = topicsDeleteDelta(followerMetadataImage.topics())
@@ -5652,11 +5657,12 @@ class ReplicaManagerTest {
// Verify that addFetcherForPartitions was called with the correct
// init offset.
+ val otherEndpoint = ClusterImageTest.IMAGE1.broker(otherId).listeners().get("PLAINTEXT")
verify(mockReplicaFetcherManager)
.addFetcherForPartitions(
Map(topicPartition -> InitialFetchState(
topicId = Some(FOO_UUID),
- leader = BrokerEndPoint(otherId, "localhost", 9093),
+ leader = BrokerEndPoint(otherId, otherEndpoint.host(), otherEndpoint.port()),
currentLeaderEpoch = 0,
initOffset = 0
))
@@ -5700,7 +5706,7 @@ class ReplicaManagerTest {
.addFetcherForPartitions(
Map(topicPartition -> InitialFetchState(
topicId = Some(FOO_UUID),
- leader = BrokerEndPoint(otherId, "localhost", 9093),
+ leader = BrokerEndPoint(otherId, otherEndpoint.host(), otherEndpoint.port()),
currentLeaderEpoch = 1,
initOffset = 1
))
@@ -5857,10 +5863,11 @@ class ReplicaManagerTest {
}
// Verify that the partition was removed and added back.
+ val localIdPlus1Endpoint = ClusterImageTest.IMAGE1.broker(localId + 1).listeners().get("PLAINTEXT")
verify(mockReplicaFetcherManager).removeFetcherForPartitions(Set(topicPartition))
verify(mockReplicaFetcherManager).addFetcherForPartitions(Map(topicPartition -> InitialFetchState(
topicId = Some(FOO_UUID),
- leader = BrokerEndPoint(localId + 1, "localhost", 9093),
+ leader = BrokerEndPoint(localId + 1, localIdPlus1Endpoint.host(), localIdPlus1Endpoint.port()),
currentLeaderEpoch = 0,
initOffset = 0
)))
@@ -5916,10 +5923,11 @@ class ReplicaManagerTest {
}
// Verify that the partition was removed and added back.
+ val localIdPlus2Endpoint = ClusterImageTest.IMAGE1.broker(localId + 2).listeners().get("PLAINTEXT")
verify(mockReplicaFetcherManager).removeFetcherForPartitions(Set(topicPartition))
verify(mockReplicaFetcherManager).addFetcherForPartitions(Map(topicPartition -> InitialFetchState(
topicId = Some(FOO_UUID),
- leader = BrokerEndPoint(localId + 2, "localhost", 9093),
+ leader = BrokerEndPoint(localId + 2, localIdPlus2Endpoint.host(), localIdPlus2Endpoint.port()),
currentLeaderEpoch = 1,
initOffset = 0
)))