Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions core/src/main/scala/kafka/admin/TopicCommand.scala
Original file line number Diff line number Diff line change
Expand Up @@ -216,7 +216,7 @@ object TopicCommand extends Logging {
def alterTopic(opts: TopicCommandOptions): Unit
def describeTopic(opts: TopicCommandOptions): Unit
def deleteTopic(opts: TopicCommandOptions): Unit
def getTopics(topicWhitelist: Option[String], excludeInternalTopics: Boolean = false): Seq[String]
def getTopics(topicIncludelist: Option[String], excludeInternalTopics: Boolean = false): Seq[String]
}

object AdminClientTopicService {
Expand Down Expand Up @@ -359,13 +359,13 @@ object TopicCommand extends Logging {
.all().get()
}

override def getTopics(topicWhitelist: Option[String], excludeInternalTopics: Boolean = false): Seq[String] = {
override def getTopics(topicIncludelist: Option[String], excludeInternalTopics: Boolean = false): Seq[String] = {
val allTopics = if (excludeInternalTopics) {
adminClient.listTopics()
} else {
adminClient.listTopics(new ListTopicsOptions().listInternal(true))
}
doGetTopics(allTopics.names().get().asScala.toSeq.sorted, topicWhitelist, excludeInternalTopics)
doGetTopics(allTopics.names().get().asScala.toSeq.sorted, topicIncludelist, excludeInternalTopics)
}

override def close(): Unit = adminClient.close()
Expand Down Expand Up @@ -515,9 +515,9 @@ object TopicCommand extends Logging {
}
}

override def getTopics(topicWhitelist: Option[String], excludeInternalTopics: Boolean = false): Seq[String] = {
override def getTopics(topicIncludelist: Option[String], excludeInternalTopics: Boolean = false): Seq[String] = {
val allTopics = zkClient.getAllTopicsInCluster().toSeq.sorted
doGetTopics(allTopics, topicWhitelist, excludeInternalTopics)
doGetTopics(allTopics, topicIncludelist, excludeInternalTopics)
}

override def close(): Unit = zkClient.close()
Expand All @@ -540,9 +540,9 @@ object TopicCommand extends Logging {
}
}

private def doGetTopics(allTopics: Seq[String], topicWhitelist: Option[String], excludeInternalTopics: Boolean): Seq[String] = {
if (topicWhitelist.isDefined) {
val topicsFilter = Whitelist(topicWhitelist.get)
private def doGetTopics(allTopics: Seq[String], topicIncludeList: Option[String], excludeInternalTopics: Boolean): Seq[String] = {
if (topicIncludeList.isDefined) {
val topicsFilter = IncludeList(topicIncludeList.get)
allTopics.filter(topicsFilter.isTopicAllowed(_, excludeInternalTopics))
} else
allTopics.filterNot(Topic.isInternal(_) && excludeInternalTopics)
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/tools/MirrorMaker.scala
Original file line number Diff line number Diff line change
Expand Up @@ -306,7 +306,7 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
val consumerRebalanceListener = new InternalRebalanceListener(this, customRebalanceListener)
whitelistOpt.foreach { whitelist =>
try {
consumer.subscribe(Pattern.compile(Whitelist(whitelist).regex), consumerRebalanceListener)
consumer.subscribe(Pattern.compile(IncludeList(whitelist).regex), consumerRebalanceListener)
} catch {
case pse: RuntimeException =>
error(s"Invalid expression syntax: $whitelist")
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import java.util.{Date, Optional, Properties}

import joptsimple.OptionParser
import kafka.api._
import kafka.utils.Whitelist
import kafka.utils.IncludeList
import kafka.utils._
import org.apache.kafka.clients._
import org.apache.kafka.clients.admin.{Admin, ListTopicsOptions, TopicDescription}
Expand Down Expand Up @@ -121,7 +121,7 @@ object ReplicaVerificationTool extends Logging {
CommandLineUtils.checkRequiredArgs(parser, options, brokerListOpt)

val regex = options.valueOf(topicWhiteListOpt)
val topicWhiteListFiler = new Whitelist(regex)
val topicWhiteListFiler = new IncludeList(regex)

try Pattern.compile(regex)
catch {
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/utils/TopicFilter.scala
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ sealed abstract class TopicFilter(rawRegex: String) extends Logging {
def isTopicAllowed(topic: String, excludeInternalTopics: Boolean): Boolean
}

case class Whitelist(rawRegex: String) extends TopicFilter(rawRegex) {
case class IncludeList(rawRegex: String) extends TopicFilter(rawRegex) {
override def isTopicAllowed(topic: String, excludeInternalTopics: Boolean) = {
val allowed = topic.matches(regex) && !(Topic.isInternal(topic) && excludeInternalTopics)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -337,7 +337,7 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
}

@Test
def testListTopicsWithWhitelist(): Unit = {
def testListTopicsWithIncludeList(): Unit = {
val topic1 = "kafka.testTopic1"
val topic2 = "kafka.testTopic2"
val topic3 = "oooof.testTopic1"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ class TopicCommandWithZKClientTest extends ZooKeeperTestHarness with Logging wit
}

@Test
def testListTopicsWithWhitelist(): Unit = {
def testListTopicsWithIncludeList(): Unit = {
val brokers = List(0, 1, 2)
TestUtils.createBrokersInZk(zkClient, brokers)

Expand Down
22 changes: 11 additions & 11 deletions core/src/test/scala/unit/kafka/utils/TopicFilterTest.scala
Original file line number Diff line number Diff line change
Expand Up @@ -24,24 +24,24 @@ import org.junit.Test
class TopicFilterTest {

@Test
def testWhitelists(): Unit = {
def testIncludeLists(): Unit = {

val topicFilter1 = Whitelist("white1,white2")
assertTrue(topicFilter1.isTopicAllowed("white2", excludeInternalTopics = true))
assertTrue(topicFilter1.isTopicAllowed("white2", excludeInternalTopics = false))
assertFalse(topicFilter1.isTopicAllowed("black1", excludeInternalTopics = true))
assertFalse(topicFilter1.isTopicAllowed("black1", excludeInternalTopics = false))
val topicFilter1 = IncludeList("yes1,yes2")
assertTrue(topicFilter1.isTopicAllowed("yes2", excludeInternalTopics = true))
assertTrue(topicFilter1.isTopicAllowed("yes2", excludeInternalTopics = false))
assertFalse(topicFilter1.isTopicAllowed("no1", excludeInternalTopics = true))
assertFalse(topicFilter1.isTopicAllowed("no1", excludeInternalTopics = false))

val topicFilter2 = Whitelist(".+")
val topicFilter2 = IncludeList(".+")
assertTrue(topicFilter2.isTopicAllowed("alltopics", excludeInternalTopics = true))
assertFalse(topicFilter2.isTopicAllowed(Topic.GROUP_METADATA_TOPIC_NAME, excludeInternalTopics = true))
assertTrue(topicFilter2.isTopicAllowed(Topic.GROUP_METADATA_TOPIC_NAME, excludeInternalTopics = false))

val topicFilter3 = Whitelist("white_listed-topic.+")
assertTrue(topicFilter3.isTopicAllowed("white_listed-topic1", excludeInternalTopics = true))
assertFalse(topicFilter3.isTopicAllowed("black1", excludeInternalTopics = true))
val topicFilter3 = IncludeList("included-topic.+")
assertTrue(topicFilter3.isTopicAllowed("included-topic1", excludeInternalTopics = true))
assertFalse(topicFilter3.isTopicAllowed("no1", excludeInternalTopics = true))

val topicFilter4 = Whitelist("test-(?!bad\\b)[\\w]+")
val topicFilter4 = IncludeList("test-(?!bad\\b)[\\w]+")
assertTrue(topicFilter4.isTopicAllowed("test-good", excludeInternalTopics = true))
assertFalse(topicFilter4.isTopicAllowed("test-bad", excludeInternalTopics = true))
}
Expand Down