def get_watermark_for_regex(
    kafka_client,
    topic_regex,
):
    """This method:
        * refreshes metadata for the kafka client
        * fetches watermarks

    :param kafka_client: KafkaToolClient instance
    :param topic: the topic regex
    :returns: dict <topic>: [ConsumerPartitionOffsets]
    """
    # Refresh client metadata. We do not use the topic list, because we
    # don't want to accidentally create the topic if it does not exist.
    # If Kafka is unavailable, let's retry loading client metadata
    try:
        kafka_client.load_metadata_for_topics()
    except KafkaUnavailableError:
        kafka_client.load_metadata_for_topics()

    topics_to_be_considered = []

    for topic in kafka_client.topic_partitions:
        if re.search(topic_regex, topic):
            topics_to_be_considered.append(topic)

    watermarks = get_topics_watermarks(
        kafka_client, topics_to_be_considered
    )
    return watermarks
This method:
        * refreshes metadata for the kafka client
        * fetches watermarks

    :param kafka_client: KafkaToolClient instance
    :param topic: the topic regex
    :returns: dict <topic>: [ConsumerPartitionOffsets]