public void resetToDatetime(final Consumer<byte[], byte[]> client,
                                final Set<TopicPartition> inputTopicPartitions,
                                final Long timestamp) {
        final Map<TopicPartition, Long> topicPartitionsAndTimes = new HashMap<>(inputTopicPartitions.size());
        for (final TopicPartition topicPartition : inputTopicPartitions) {
            topicPartitionsAndTimes.put(topicPartition, timestamp);
        }

        final Map<TopicPartition, OffsetAndTimestamp> topicPartitionsAndOffset = client.offsetsForTimes(topicPartitionsAndTimes);

        for (final TopicPartition topicPartition : inputTopicPartitions) {
            final Optional<Long> partitionOffset = Optional.ofNullable(topicPartitionsAndOffset.get(topicPartition))
                    .map(OffsetAndTimestamp::offset)
                    .filter(offset -> offset != ListOffsetsResponse.UNKNOWN_OFFSET);
            if (partitionOffset.isPresent()) {
                client.seek(topicPartition, partitionOffset.get());
            } else {
                client.seekToEnd(Collections.singletonList(topicPartition));
                System.out.println("Partition " + topicPartition.partition() + " from topic " + topicPartition.topic() +
                        " is empty, without a committed record. Falling back to latest known offset.");
            }
        }
    }
@Test
    public void testResetToDatetimeWhenPartitionIsEmptyResetsToLatestOffset() {
        final long beginningAndEndOffset = 5L; // Empty partition implies beginning offset == end offset
        final MockConsumer<byte[], byte[]> emptyConsumer = new EmptyPartitionConsumer<>(OffsetResetStrategy.EARLIEST);
        emptyConsumer.assign(Collections.singletonList(topicPartition));

        final Map<TopicPartition, Long> beginningOffsetsMap = new HashMap<>();
        beginningOffsetsMap.put(topicPartition, beginningAndEndOffset);
        emptyConsumer.updateBeginningOffsets(beginningOffsetsMap);

        final Map<TopicPartition, Long> endOffsetsMap = new HashMap<>();
        endOffsetsMap.put(topicPartition, beginningAndEndOffset);
        emptyConsumer.updateEndOffsets(endOffsetsMap);

        final long yesterdayTimestamp = Instant.now().minus(Duration.ofDays(1)).toEpochMilli();
        // resetToDatetime only seeks the offset, but does not commit.
        streamsResetter.resetToDatetime(emptyConsumer, inputTopicPartitions, yesterdayTimestamp);

        final long position = emptyConsumer.position(topicPartition);

        assertEquals(beginningAndEndOffset, position);
    }