You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dimitarndimitrov (via GitHub)" <gi...@apache.org> on 2025/01/21 15:10:40 UTC
[PR] Cherry-pick KAFKA-18583 fix (#18635) to 3.9 [kafka]
dimitarndimitrov opened a new pull request, #18657:
URL: https://github.com/apache/kafka/pull/18657
The cherry-pick required reimplementing the accompanying test to work with `UpdateMetadataRequest` (removed in 4.0 and trunk) in order to also apply to `ZkMetadataCache`. If the removal of `UpdateMetadataRequest` is backported here as well, the test can be changed to match the trunk version.
- The test now passes locally against both the ZK and the KRaft metadata caches.
_Conflicts: core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala_
### 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] Cherry-pick KAFKA-18583 fix (#18635) to 3.9 [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #18657:
URL: https://github.com/apache/kafka/pull/18657#issuecomment-2606611401
Merged to 3.9 and 3.8.
--
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] Cherry-pick KAFKA-18583 fix (#18635) to 3.9 [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18657:
URL: https://github.com/apache/kafka/pull/18657#issuecomment-2605138101
> If the removal of UpdateMetadataRequest is backported here as well, the test can be changed to match the trunk version.
That is 4.0+ only, so we don't have to worry about that.
--
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] Cherry-pick KAFKA-18583 fix (#18635) to 3.9 [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #18657:
URL: https://github.com/apache/kafka/pull/18657
--
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] Cherry-pick KAFKA-18583 fix (#18635) to 3.9 [kafka]
Posted by "dimitarndimitrov (via GitHub)" <gi...@apache.org>.
dimitarndimitrov commented on code in PR #18657:
URL: https://github.com/apache/kafka/pull/18657#discussion_r1925019540
##########
core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala:
##########
@@ -645,6 +646,114 @@ class MetadataCacheTest {
assertEquals(Seq(expectedNode1), partitionInfo.offlineReplicas.toSeq)
}
+ @ParameterizedTest
+ @MethodSource(Array("cacheProvider"))
+ def testGetPartitionReplicaEndpoints(cache: MetadataCache): Unit = {
+ val securityProtocol = SecurityProtocol.PLAINTEXT
+ val listenerName = ListenerName.forSecurityProtocol(securityProtocol)
+
+ // Set up broker data for the metadata cache
+ val numBrokers = 10
+ // Set only the last broker in the list to be offline in order to allow easy
+ // indexing of brokers in the brokerStates list - the index in the list will
+ // be the same as the brokerId of the broker at that position.
+ val offlineBrokerId = numBrokers - 1
+ val brokerStates = (0 until numBrokers - 1).map { brokerId =>
+ new UpdateMetadataBroker()
+ .setId(brokerId)
+ .setRack("rack" + (brokerId % 3))
+ .setEndpoints(
+ Seq(new UpdateMetadataEndpoint()
+ .setHost("foo" + brokerId)
+ .setPort(9092)
+ .setSecurityProtocol(securityProtocol.id)
+ .setListener(listenerName.value)
+ ).asJava)
+ }
+
+ val topic = "many-partitions-topic"
+ val topicId = Uuid.randomUuid()
+
+ // Set up a number of partitions such that each different combination of
+ // $replicationFactor brokers is made a replica set for exactly one partition
+ val replicationFactor = 3
+ val replicaSets = getAllReplicaSets(numBrokers, replicationFactor)
+ val numPartitions = replicaSets.length
+ val partitionStates = (0 until numPartitions).map { partitionId =>
+ val replicas = replicaSets(partitionId)
+ val onlineReplicas = replicas.stream().filter(id => id != offlineBrokerId).collect(Collectors.toList())
+ new UpdateMetadataPartitionState()
+ .setTopicName(topic)
+ .setPartitionIndex(partitionId)
+ .setReplicas(replicas)
+ .setLeader(onlineReplicas.get(0))
+ .setIsr(onlineReplicas)
+ .setOfflineReplicas(Collections.singletonList(offlineBrokerId))
+ }
+
+ // Load the prepared data in the metadata cache
+ val version = ApiKeys.UPDATE_METADATA.latestVersion
+ val controllerId = 0
+ val controllerEpoch = 123
+ val updateMetadataRequest = new UpdateMetadataRequest.Builder(
+ version,
+ controllerId,
+ controllerEpoch,
+ brokerEpoch,
+ partitionStates.asJava,
+ brokerStates.asJava,
+ Collections.singletonMap(topic, topicId)).build()
+ MetadataCacheTest.updateCache(cache, updateMetadataRequest)
+
+ (0 until numPartitions).foreach { partitionId =>
+ val tp = new TopicPartition(topic, partitionId)
+ val brokerIdToNodeMap = cache.getPartitionReplicaEndpoints(tp, listenerName)
+ val replicaSet = brokerIdToNodeMap.keySet
+ val expectedReplicaSet = partitionStates(partitionId).replicas().asScala.toSet
+ // Verify that we have endpoints for exactly the non-fenced brokers of the replica set
+ if (expectedReplicaSet.contains(offlineBrokerId)) {
+ assertEquals(expectedReplicaSet,
+ replicaSet + offlineBrokerId,
+ s"Unexpected partial replica set for partition $partitionId")
Review Comment:
nit: This condition should also have asserted that `replicaSet` doesn't contain `offlineBrokerId` - right now the existing `assertEquals` will pass even if that's not the case. Also applies to the original `trunk` change.
--
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