KAFKA-14647: Moving TopicFilter to server-common/utils

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" />
             </subpackage>

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)

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))

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))
-  }
-}

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")

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) {

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;
+        }
+    }
+}