You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2023/07/18 08:39:04 UTC
[kafka] branch trunk updated: KAFKA-14647: Move TopicFilter to server-common/utils (#13158)
This is an automated email from the ASF dual-hosted git repository.
mimaison 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 fa5b493241d KAFKA-14647: Move TopicFilter to server-common/utils (#13158)
fa5b493241d is described below
commit fa5b493241d43c6d8809853678dcd05fb8fa518a
Author: vamossagar12 <sa...@gmail.com>
AuthorDate: Tue Jul 18 14:08:56 2023 +0530
KAFKA-14647: Move TopicFilter to server-common/utils (#13158)
Reviewers: Mickael Maison <mi...@gmail.com>, Federico Valeri <fe...@gmail.com>
---
checkstyle/import-control-server-common.xml | 1 +
core/src/main/scala/kafka/admin/TopicCommand.scala | 3 +-
.../main/scala/kafka/tools/ConsoleConsumer.scala | 1 -
.../main/scala/kafka/tools/GetOffsetShell.scala | 91 +++++-----------------
core/src/main/scala/kafka/tools/MirrorMaker.scala | 4 +-
.../kafka/tools/ReplicaVerificationTool.scala | 3 +-
core/src/main/scala/kafka/utils/TopicFilter.scala | 55 -------------
.../scala/unit/kafka/utils/TopicFilterTest.scala | 49 ------------
.../apache/kafka/server/util/PartitionFilter.java | 66 ++++++++++++++++
.../org/apache/kafka/server/util/TopicFilter.java | 73 +++++++++++++++++
.../kafka/server/util/TopicPartitionFilter.java | 71 +++++++++++++++++
.../apache/kafka/server/util/TopicFilterTest.java | 51 ++++++++++++
12 files changed, 287 insertions(+), 181 deletions(-)
diff --git a/checkstyle/import-control-server-common.xml b/checkstyle/import-control-server-common.xml
index 8e96ea9390c..4002e3ea5bd 100644
--- a/checkstyle/import-control-server-common.xml
+++ b/checkstyle/import-control-server-common.xml
@@ -90,6 +90,7 @@
<allow class="org.apache.kafka.clients.KafkaClient" />
<allow class="org.apache.kafka.clients.RequestCompletionHandler" />
+ <allow class="org.apache.kafka.server.util.TopicFilter.IncludeList" />
<subpackage name="timer">
<allow class="org.apache.kafka.server.util.MockTime" />
<allow class="org.apache.kafka.server.util.ShutdownableThread" />
diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala
index f54e5720930..b4c74d35cdf 100755
--- a/core/src/main/scala/kafka/admin/TopicCommand.scala
+++ b/core/src/main/scala/kafka/admin/TopicCommand.scala
@@ -35,6 +35,7 @@ import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.server.util.{CommandDefaultOptions, CommandLineUtils}
import org.apache.kafka.storage.internals.log.LogConfig
+import org.apache.kafka.server.util.TopicFilter.IncludeList
import scala.annotation.nowarn
import scala.jdk.CollectionConverters._
@@ -419,7 +420,7 @@ object TopicCommand extends Logging {
private def doGetTopics(allTopics: Seq[String], topicIncludeList: Option[String], excludeInternalTopics: Boolean): Seq[String] = {
if (topicIncludeList.isDefined) {
- val topicsFilter = IncludeList(topicIncludeList.get)
+ val topicsFilter = new IncludeList(topicIncludeList.get)
allTopics.filter(topicsFilter.isTopicAllowed(_, excludeInternalTopics))
} else
allTopics.filterNot(Topic.isInternal(_) && excludeInternalTopics)
diff --git a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
index 2774f13397a..44a80dcd563 100755
--- a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
+++ b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
@@ -302,7 +302,6 @@ object ConsoleConsumer extends Logging {
// topic must be specified.
var topicArg: String = _
var includedTopicsArg: String = _
- var filterSpec: TopicFilter = _
val extraConsumerProps = CommandLineUtils.parseKeyValueArgs(options.valuesOf(consumerPropertyOpt))
val consumerProps = if (options.has(consumerConfigOpt))
Utils.loadProps(options.valueOf(consumerConfigOpt))
diff --git a/core/src/main/scala/kafka/tools/GetOffsetShell.scala b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
index 379b92218c8..1acbee6976d 100644
--- a/core/src/main/scala/kafka/tools/GetOffsetShell.scala
+++ b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
@@ -19,12 +19,19 @@
package kafka.tools
import joptsimple._
-import kafka.utils.{Exit, IncludeList, ToolsUtils}
+import kafka.utils.{Exit, ToolsUtils}
import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, ListTopicsOptions, OffsetSpec}
import org.apache.kafka.common.{KafkaException, TopicPartition}
import org.apache.kafka.common.requests.{ListOffsetsRequest, ListOffsetsResponse}
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.server.util.CommandLineUtils
+import org.apache.kafka.server.util.TopicFilter.IncludeList
+import org.apache.kafka.server.util.TopicPartitionFilter
+import org.apache.kafka.server.util.TopicPartitionFilter.TopicFilterAndPartitionFilter
+import org.apache.kafka.server.util.TopicPartitionFilter.CompositeTopicPartitionFilter
+import org.apache.kafka.server.util.PartitionFilter.UniquePartitionFilter
+import org.apache.kafka.server.util.PartitionFilter.PartitionRangeFilter
+import org.apache.kafka.server.util.PartitionFilter.PartitionsSetFilter
import java.util.Properties
import java.util.concurrent.ExecutionException
@@ -197,8 +204,8 @@ object GetOffsetShell {
topicPartitions: String
): TopicPartitionFilter = {
val ruleSpecs = topicPartitions.split(",")
- val rules = ruleSpecs.map(ruleSpec => parseRuleSpec(ruleSpec))
- CompositeTopicPartitionFilter(rules)
+ val rules = ruleSpecs.toSeq.map(ruleSpec => parseRuleSpec(ruleSpec))
+ new CompositeTopicPartitionFilter(rules.asJava)
}
def parseRuleSpec(ruleSpec: String): TopicPartitionFilter = {
@@ -210,16 +217,16 @@ object GetOffsetShell {
Option(matcher.group(group)).filter(s => s != null && s.nonEmpty)
}
- val topicFilter = IncludeList(group(1).getOrElse(".*"))
+ val topicFilter = new IncludeList(group(1).getOrElse(".*"))
val partitionFilter = group(2).map(_.toInt) match {
case Some(partition) =>
- UniquePartitionFilter(partition)
+ new UniquePartitionFilter(partition)
case None =>
val lowerRange = group(3).map(_.toInt).getOrElse(0)
val upperRange = group(4).map(_.toInt).getOrElse(Int.MaxValue)
- PartitionRangeFilter(lowerRange, upperRange)
+ new PartitionRangeFilter(lowerRange, upperRange)
}
- TopicFilterAndPartitionFilter(
+ new TopicFilterAndPartitionFilter(
topicFilter,
partitionFilter
)
@@ -232,18 +239,18 @@ object GetOffsetShell {
topicOpt: Option[String],
partitionIds: String
): TopicFilterAndPartitionFilter = {
- TopicFilterAndPartitionFilter(
- IncludeList(topicOpt.getOrElse(".*")),
- PartitionsSetFilter(createPartitionSet(partitionIds))
+ new TopicFilterAndPartitionFilter(
+ new IncludeList(topicOpt.getOrElse(".*")),
+ new PartitionsSetFilter(createPartitionSet(partitionIds).asJava)
)
}
- def createPartitionSet(partitionsString: String): Set[Int] = {
+ def createPartitionSet(partitionsString: String): Set[Integer] = {
if (partitionsString == null || partitionsString.isEmpty)
Set.empty
else
partitionsString.split(",").map { partitionString =>
- try partitionString.toInt
+ try Integer.valueOf(partitionString)
catch {
case _: NumberFormatException =>
throw new IllegalArgumentException(s"--partitions expects a comma separated list of numeric " +
@@ -274,63 +281,3 @@ object GetOffsetShell {
}
}
-trait PartitionFilter {
-
- /**
- * Used to filter partitions based on a certain criteria, for example, a set of partition ids.
- */
- def isPartitionAllowed(partition: Int): Boolean
-}
-
-case class PartitionsSetFilter(partitionIds: Set[Int]) extends PartitionFilter {
- override def isPartitionAllowed(partition: Int): Boolean = partitionIds.isEmpty || partitionIds.contains(partition)
-}
-
-case class UniquePartitionFilter(partition: Int) extends PartitionFilter {
- override def isPartitionAllowed(partition: Int): Boolean = partition == this.partition
-}
-
-case class PartitionRangeFilter(lowerRange: Int, upperRange: Int) extends PartitionFilter {
- override def isPartitionAllowed(partition: Int): Boolean = partition >= lowerRange && partition < upperRange
-}
-
-trait TopicPartitionFilter {
-
- /**
- * Used to filter topics based on a certain criteria, for example, a set of topic names or a regular expression.
- */
- def isTopicAllowed(topic: String): Boolean
-
- /**
- * Used to filter topic-partitions based on a certain criteria, for example, a topic pattern and a set of partition ids.
- */
- def isTopicPartitionAllowed(partition: TopicPartition): Boolean
-}
-
-/**
- * Creates a topic-partition filter based on a topic filter and a partition filter
- */
-case class TopicFilterAndPartitionFilter(
- topicFilter: IncludeList,
- partitionFilter: PartitionFilter
-) extends TopicPartitionFilter {
-
- override def isTopicPartitionAllowed(partition: TopicPartition): Boolean = {
- isTopicAllowed(partition.topic) && partitionFilter.isPartitionAllowed(partition.partition)
- }
-
- override def isTopicAllowed(topic: String): Boolean = {
- topicFilter.isTopicAllowed(topic, false)
- }
-}
-
-case class CompositeTopicPartitionFilter(filters: Array[TopicPartitionFilter]) extends TopicPartitionFilter {
-
- override def isTopicAllowed(topic: String): Boolean = {
- filters.exists(_.isTopicAllowed(topic))
- }
-
- override def isTopicPartitionAllowed(tp: TopicPartition): Boolean = {
- filters.exists(_.isTopicPartitionAllowed(tp))
- }
-}
diff --git a/core/src/main/scala/kafka/tools/MirrorMaker.scala b/core/src/main/scala/kafka/tools/MirrorMaker.scala
index 65c1013fcbf..d7cf6dbb275 100755
--- a/core/src/main/scala/kafka/tools/MirrorMaker.scala
+++ b/core/src/main/scala/kafka/tools/MirrorMaker.scala
@@ -21,7 +21,6 @@ import java.time.Duration
import java.util
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
import java.util.concurrent.CountDownLatch
-import java.util.regex.Pattern
import java.util.{Collections, Properties}
import kafka.consumer.BaseConsumerRecord
import kafka.utils._
@@ -35,6 +34,7 @@ import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.common.{KafkaException, TopicPartition}
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.{CommandDefaultOptions, CommandLineUtils}
+import org.apache.kafka.server.util.TopicFilter.IncludeList
import scala.jdk.CollectionConverters._
import scala.collection.mutable.HashMap
@@ -309,7 +309,7 @@ object MirrorMaker extends Logging {
val consumerRebalanceListener = new InternalRebalanceListener(this, customRebalanceListener)
includeOpt.foreach { include =>
try {
- consumer.subscribe(Pattern.compile(IncludeList(include).regex), consumerRebalanceListener)
+ consumer.subscribe(new IncludeList(include).getPattern, consumerRebalanceListener)
} catch {
case pse: RuntimeException =>
error(s"Invalid expression syntax: $include")
diff --git a/core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala b/core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala
index 649fed0ae26..3f66eaa8b7d 100644
--- a/core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala
+++ b/core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala
@@ -42,6 +42,7 @@ import java.util.regex.{Pattern, PatternSyntaxException}
import java.util.{Date, Optional, Properties}
import scala.collection.Seq
import scala.jdk.CollectionConverters._
+import org.apache.kafka.server.util.TopicFilter.IncludeList
/**
* For verifying the consistency among replicas.
@@ -154,7 +155,7 @@ object ReplicaVerificationTool extends Logging {
val topicIds = topicsMetadata.map( metadata => metadata.name() -> metadata.topicId()).toMap
val filteredTopicMetadata = topicsMetadata.filter { topicMetaData =>
- topicsIncludeFilter.isTopicAllowed(topicMetaData.name, excludeInternalTopics = false)
+ topicsIncludeFilter.isTopicAllowed(topicMetaData.name, false)
}
if (filteredTopicMetadata.isEmpty) {
diff --git a/core/src/main/scala/kafka/utils/TopicFilter.scala b/core/src/main/scala/kafka/utils/TopicFilter.scala
deleted file mode 100644
index 075c14722ee..00000000000
--- a/core/src/main/scala/kafka/utils/TopicFilter.scala
+++ /dev/null
@@ -1,55 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kafka.utils
-
-import java.util.regex.{Pattern, PatternSyntaxException}
-
-import org.apache.kafka.common.internals.Topic
-
-sealed abstract class TopicFilter(rawRegex: String) extends Logging {
-
- val regex = rawRegex
- .trim
- .replace(',', '|')
- .replace(" ", "")
- .replaceAll("""^["']+""","")
- .replaceAll("""["']+$""","") // property files may bring quotes
-
- try {
- Pattern.compile(regex)
- }
- catch {
- case _: PatternSyntaxException =>
- throw new RuntimeException(regex + " is an invalid regex.")
- }
-
- override def toString = regex
-
- def isTopicAllowed(topic: String, excludeInternalTopics: Boolean): Boolean
-}
-
-case class IncludeList(rawRegex: String) extends TopicFilter(rawRegex) {
- override def isTopicAllowed(topic: String, excludeInternalTopics: Boolean) = {
- val allowed = topic.matches(regex) && !(Topic.isInternal(topic) && excludeInternalTopics)
-
- debug("%s %s".format(
- topic, if (allowed) "allowed" else "filtered"))
-
- allowed
- }
-}
diff --git a/core/src/test/scala/unit/kafka/utils/TopicFilterTest.scala b/core/src/test/scala/unit/kafka/utils/TopicFilterTest.scala
deleted file mode 100644
index ea728dced74..00000000000
--- a/core/src/test/scala/unit/kafka/utils/TopicFilterTest.scala
+++ /dev/null
@@ -1,49 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kafka.utils
-
-import org.apache.kafka.common.internals.Topic
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.Test
-
-class TopicFilterTest {
-
- @Test
- def testIncludeLists(): Unit = {
-
- 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 = 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 = IncludeList("included-topic.+")
- assertTrue(topicFilter3.isTopicAllowed("included-topic1", excludeInternalTopics = true))
- assertFalse(topicFilter3.isTopicAllowed("no1", excludeInternalTopics = true))
-
- val topicFilter4 = IncludeList("test-(?!bad\\b)[\\w]+")
- assertTrue(topicFilter4.isTopicAllowed("test-good", excludeInternalTopics = true))
- assertFalse(topicFilter4.isTopicAllowed("test-bad", excludeInternalTopics = true))
- }
-
-}
diff --git a/server-common/src/main/java/org/apache/kafka/server/util/PartitionFilter.java b/server-common/src/main/java/org/apache/kafka/server/util/PartitionFilter.java
new file mode 100644
index 00000000000..d66ae4776ce
--- /dev/null
+++ b/server-common/src/main/java/org/apache/kafka/server/util/PartitionFilter.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.server.util;
+
+import java.util.Set;
+
+public interface PartitionFilter {
+ /**
+ * Used to filter partitions based on a certain criteria, for example, a set of partition ids.
+ */
+ boolean isPartitionAllowed(int partition);
+
+ class PartitionsSetFilter implements PartitionFilter {
+ private final Set<Integer> partitionIds;
+ public PartitionsSetFilter(Set<Integer> partitionIds) {
+ this.partitionIds = partitionIds;
+ }
+ @Override
+ public boolean isPartitionAllowed(int partition) {
+ return partitionIds.isEmpty() || partitionIds.contains(partition);
+ }
+ }
+
+ class UniquePartitionFilter implements PartitionFilter {
+ private final int partition;
+
+ public UniquePartitionFilter(int partition) {
+ this.partition = partition;
+ }
+
+ @Override
+ public boolean isPartitionAllowed(int partition) {
+ return partition == this.partition;
+ }
+ }
+
+ class PartitionRangeFilter implements PartitionFilter {
+ private final int lowerRange;
+ private final int upperRange;
+
+ public PartitionRangeFilter(int lowerRange, int upperRange) {
+ this.lowerRange = lowerRange;
+ this.upperRange = upperRange;
+ }
+
+ @Override
+ public boolean isPartitionAllowed(int partition) {
+ return partition >= lowerRange && partition < upperRange;
+ }
+ }
+}
diff --git a/server-common/src/main/java/org/apache/kafka/server/util/TopicFilter.java b/server-common/src/main/java/org/apache/kafka/server/util/TopicFilter.java
new file mode 100644
index 00000000000..4127ce46d82
--- /dev/null
+++ b/server-common/src/main/java/org/apache/kafka/server/util/TopicFilter.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.util;
+
+import org.apache.kafka.common.internals.Topic;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+public abstract class TopicFilter {
+
+ protected final String regex;
+ private final Pattern pattern;
+ public TopicFilter(String rawRegex) {
+ this.regex = rawRegex
+ .trim()
+ .replace(',', '|')
+ .replace(" ", "")
+ .replaceAll("^[\"']+", "")
+ .replaceAll("[\"']+$", ""); // property files may bring quotes
+ try {
+ this.pattern = Pattern.compile(regex);
+ } catch (PatternSyntaxException e) {
+ throw new RuntimeException(regex + " is an invalid regex.");
+ }
+ }
+
+ public abstract boolean isTopicAllowed(String topic, boolean excludeInternalTopics);
+
+ @Override
+ public String toString() {
+ return this.regex;
+ }
+
+ public Pattern getPattern() {
+ return pattern;
+ }
+
+ public static class IncludeList extends TopicFilter {
+ private final Logger log = LoggerFactory.getLogger(IncludeList.class);
+ public IncludeList(String rawRegex) {
+ super(rawRegex);
+ }
+
+ @Override
+ public boolean isTopicAllowed(String topic, boolean excludeInternalTopics) {
+ boolean allowed = topic.matches(regex) && !(Topic.isInternal(topic) && excludeInternalTopics);
+ if (allowed) {
+ log.debug("{} allowed", topic);
+ } else {
+ log.debug("{} filtered", topic);
+ }
+ return allowed;
+ }
+ }
+
+}
diff --git a/server-common/src/main/java/org/apache/kafka/server/util/TopicPartitionFilter.java b/server-common/src/main/java/org/apache/kafka/server/util/TopicPartitionFilter.java
new file mode 100644
index 00000000000..d232da6898e
--- /dev/null
+++ b/server-common/src/main/java/org/apache/kafka/server/util/TopicPartitionFilter.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.util;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.server.util.TopicFilter.IncludeList;
+
+import java.util.List;
+
+public interface TopicPartitionFilter {
+ /**
+ * Used to filter topics based on a certain criteria, for example, a set of topic names or a regular expression.
+ */
+ boolean isTopicAllowed(String topic);
+
+ /**
+ * Used to filter topic-partitions based on a certain criteria, for example, a topic pattern and a set of partition ids.
+ */
+ boolean isTopicPartitionAllowed(TopicPartition partition);
+
+ class TopicFilterAndPartitionFilter implements TopicPartitionFilter {
+ private final IncludeList topicFilter;
+ private final PartitionFilter partitionFilter;
+ public TopicFilterAndPartitionFilter(IncludeList topicFilter, PartitionFilter partitionFilter) {
+ this.topicFilter = topicFilter;
+ this.partitionFilter = partitionFilter;
+ }
+
+ @Override
+ public boolean isTopicAllowed(String topic) {
+ return topicFilter.isTopicAllowed(topic, false);
+ }
+
+ @Override
+ public boolean isTopicPartitionAllowed(TopicPartition partition) {
+ return isTopicAllowed(partition.topic()) && partitionFilter.isPartitionAllowed(partition.partition());
+ }
+ }
+
+ class CompositeTopicPartitionFilter implements TopicPartitionFilter {
+ private final List<TopicPartitionFilter> filters;
+
+ public CompositeTopicPartitionFilter(List<TopicPartitionFilter> filters) {
+ this.filters = filters;
+ }
+
+ @Override
+ public boolean isTopicAllowed(String topic) {
+ return filters.stream().anyMatch(tp -> tp.isTopicAllowed(topic));
+ }
+
+ @Override
+ public boolean isTopicPartitionAllowed(TopicPartition partition) {
+ return filters.stream().anyMatch(tp -> tp.isTopicPartitionAllowed(partition));
+ }
+ }
+}
diff --git a/server-common/src/test/java/org/apache/kafka/server/util/TopicFilterTest.java b/server-common/src/test/java/org/apache/kafka/server/util/TopicFilterTest.java
new file mode 100644
index 00000000000..634bae7e74f
--- /dev/null
+++ b/server-common/src/test/java/org/apache/kafka/server/util/TopicFilterTest.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.util;
+
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.server.util.TopicFilter.IncludeList;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+public class TopicFilterTest {
+ @Test
+ public void testIncludeLists() {
+ IncludeList topicFilter1 = new TopicFilter.IncludeList("yes1,yes2");
+ assertTrue(topicFilter1.isTopicAllowed("yes2", true));
+ assertTrue(topicFilter1.isTopicAllowed("yes2", false));
+ assertFalse(topicFilter1.isTopicAllowed("no1", true));
+ assertFalse(topicFilter1.isTopicAllowed("no1", false));
+
+ IncludeList topicFilter2 = new IncludeList(".+");
+ assertTrue(topicFilter2.isTopicAllowed("alltopics", true));
+ assertFalse(topicFilter2.isTopicAllowed(Topic.GROUP_METADATA_TOPIC_NAME, true));
+ assertTrue(topicFilter2.isTopicAllowed(Topic.GROUP_METADATA_TOPIC_NAME, false));
+
+ assertFalse(topicFilter2.isTopicAllowed(Topic.TRANSACTION_STATE_TOPIC_NAME, true));
+ assertTrue(topicFilter2.isTopicAllowed(Topic.TRANSACTION_STATE_TOPIC_NAME, false));
+
+ IncludeList topicFilter3 = new IncludeList("included-topic.+");
+ assertTrue(topicFilter3.isTopicAllowed("included-topic1", true));
+ assertFalse(topicFilter3.isTopicAllowed("no1", true));
+
+ IncludeList topicFilter4 = new IncludeList("test-(?!bad\\b)[\\w]+");
+ assertTrue(topicFilter4.isTopicAllowed("test-good", true));
+ assertFalse(topicFilter4.isTopicAllowed("test-bad", true));
+ }
+}