@Override
    protected Map<KafkaTopicPartition, Long> fetchOffsetsWithTimestamp(
            Collection<KafkaTopicPartition> partitions,
            long timestamp) {

        Map<TopicPartition, Long> partitionOffsetsRequest = new HashMap<>(partitions.size());
        for (KafkaTopicPartition partition : partitions) {
            partitionOffsetsRequest.put(
                    new TopicPartition(partition.getTopic(), partition.getPartition()),
                    timestamp);
        }

        final Map<KafkaTopicPartition, Long> result = new HashMap<>(partitions.size());
        // use a short-lived consumer to fetch the offsets;
        // this is ok because this is a one-time operation that happens only on startup
        try (KafkaConsumer<?, ?> consumer = new KafkaConsumer(properties)) {
            for (Map.Entry<TopicPartition, OffsetAndTimestamp> partitionToOffset :
                    consumer.offsetsForTimes(partitionOffsetsRequest).entrySet()) {

                result.put(
                        new KafkaTopicPartition(partitionToOffset.getKey().topic(), partitionToOffset.getKey().partition()),
                        (partitionToOffset.getValue() == null) ? null : partitionToOffset.getValue().offset());
            }

        }
        return result;
    }
Creates a new Kafka streaming source consumer. Use this constructor to
subscribe to multiple topics based on a regular expression pattern.

<p>If partition discovery is enabled (by setting a non-negative value for
{@link FlinkKafkaConsumer#KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS} in the properties), topics
with names matching the pattern will also be subscribed to as they are created on the fly.

<p>This constructor allows passing a {@see KafkaDeserializationSchema} for reading key/value
pairs, offsets, and topic names from Kafka.

@param subscriptionPattern The regular expression for a pattern of topic names to subscribe to.
@param deserializer        The keyed de-/serializer used to convert between Kafka's byte messages and Flink's objects.
@param props
kafka/kafka-source/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java