dajac commented on a change in pull request #11170:
URL: https://github.com/apache/kafka/pull/11170#discussion_r714936472
##########
File path: core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala
##########
@@ -929,6 +929,29 @@ class AbstractFetcherThreadTest {
fetcher.verifyLastFetchedEpoch(partition, Some(5))
}
+ @Test
+ def testMaybeUpdateTopicIds(): Unit = {
+ val partition = new TopicPartition("topic1", 0)
+ val fetcher = new MockFetcherThread
+
+ // Start with no topic IDs
+ fetcher.setReplicaState(partition,
MockFetcherThread.PartitionState(leaderEpoch = 0))
+ fetcher.addPartitions(Map(partition -> initialFetchState(None, 0L,
leaderEpoch = 0)))
+
+
+ def verifyFetchState(fetchState: Option[PartitionFetchState],
expectedTopicId: Option[Uuid]): Unit = {
+ assertTrue(fetchState.isDefined)
+ assertEquals(expectedTopicId, fetchState.get.topicId)
+ }
+
+ verifyFetchState(fetcher.fetchState(partition), None)
+
+ // Add topic ID
+ fetcher.maybeUpdateTopicIds(Set(partition), topicName =>
topicIds.get(topicName))
Review comment:
We could add tests for the ReplicaManager which verifies that the topic
id is updated and propagated to the fetcher manager. Is it what you are
thinking about?
Regarding this particular test, what would you add? If the topic id is zero,
we would just set it. As you said, the logic which prevents this from happening
is before.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
