diff --git a/checkstyle/import-control-server-common.xml b/checkstyle/import-control-server-common.xml index 2e65ec601a395..07638af32defc 100644 --- a/checkstyle/import-control-server-common.xml +++ b/checkstyle/import-control-server-common.xml @@ -90,6 +90,7 @@ + diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index f54e572093026..b4c74d35cdfc9 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 2774f13397a62..44a80dcd563ae 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 379b92218c80a..1acbee6976d56 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 65c1013fcbf48..d7cf6dbb275ce 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 649fed0ae2694..3f66eaa8b7ddd 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 075c14722ee96..0000000000000 --- 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 ea728dced74b4..0000000000000 --- 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 0000000000000..d66ae4776ce25 --- /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 partitionIds; + public PartitionsSetFilter(Set 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 0000000000000..4127ce46d8200 --- /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 0000000000000..d232da6898eec --- /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 filters; + + public CompositeTopicPartitionFilter(List 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 0000000000000..634bae7e74f97 --- /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)); + } +}