static Collection<TopicPartition> getMatchingTopicPartitions(
        Admin adminClient, String topicRegex, int startPartition, int endPartition)
        throws Throwable {
        final Pattern topicNamePattern = Pattern.compile(topicRegex);

        // first get list of matching topics
        List<String> matchedTopics = new ArrayList<>();
        ListTopicsResult res = adminClient.listTopics(
            new ListTopicsOptions().timeoutMs(ADMIN_REQUEST_TIMEOUT));
        Map<String, TopicListing> topicListingMap = res.namesToListings().get();
        for (Map.Entry<String, TopicListing> topicListingEntry: topicListingMap.entrySet()) {
            if (!topicListingEntry.getValue().isInternal()
                && topicNamePattern.matcher(topicListingEntry.getKey()).matches()) {
                matchedTopics.add(topicListingEntry.getKey());
            }
        }

        // create a list of topic/partitions
        List<TopicPartition> out = new ArrayList<>();
        DescribeTopicsResult topicsResult = adminClient.describeTopics(
            matchedTopics, new DescribeTopicsOptions().timeoutMs(ADMIN_REQUEST_TIMEOUT));
        Map<String, TopicDescription> topicDescriptionMap = topicsResult.allTopicNames().get();
        for (TopicDescription desc: topicDescriptionMap.values()) {
            List<TopicPartitionInfo> partitions = desc.partitions();
            for (TopicPartitionInfo info: partitions) {
                if ((info.partition() >= startPartition) && (info.partition() <= endPartition)) {
                    out.add(new TopicPartition(desc.name(), info.partition()));
                }
            }
        }
        return out;
    }
@Test
    public void testGetMatchingTopicPartitionsCorrectlyMatchesTopics() throws Throwable {
        final String topic1 = "test-topic";
        final String topic2 = "another-test-topic";
        final String topic3 = "one-more";
        makeExistingTopicWithOneReplica(topic1, 10);
        makeExistingTopicWithOneReplica(topic2, 20);
        makeExistingTopicWithOneReplica(topic3, 30);

        Collection<TopicPartition> topicPartitions =
            WorkerUtils.getMatchingTopicPartitions(adminClient, ".*-topic$", 0, 1);
        assertEquals(
            Utils.mkSet(
                new TopicPartition(topic1, 0), new TopicPartition(topic1, 1),
                new TopicPartition(topic2, 0), new TopicPartition(topic2, 1)
            ),
            new HashSet<>(topicPartitions)
        );
    }