public void resetOffsetsTo(final Consumer<byte[], byte[]> client,
                               final Set<TopicPartition> inputTopicPartitions,
                               final Long offset) {
        final Map<TopicPartition, Long> endOffsets = client.endOffsets(inputTopicPartitions);
        final Map<TopicPartition, Long> beginningOffsets = client.beginningOffsets(inputTopicPartitions);

        final Map<TopicPartition, Long> topicPartitionsAndOffset = new HashMap<>(inputTopicPartitions.size());
        for (final TopicPartition topicPartition : inputTopicPartitions) {
            topicPartitionsAndOffset.put(topicPartition, offset);
        }

        final Map<TopicPartition, Long> validatedTopicPartitionsAndOffset =
            checkOffsetRange(topicPartitionsAndOffset, beginningOffsets, endOffsets);

        for (final TopicPartition topicPartition : inputTopicPartitions) {
            client.seek(topicPartition, validatedTopicPartitionsAndOffset.get(topicPartition));
        }
    }
@Test
    public void testResetOffsetToSpecificOffsetWhenAfterEndOffset() {
        final long beginningOffset = 5L;
        final long endOffset = 10L;
        final MockConsumer<byte[], byte[]> emptyConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
        emptyConsumer.assign(Collections.singletonList(topicPartition));

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

        final Map<TopicPartition, Long> endOffsetsMap = new HashMap<>();
        endOffsetsMap.put(topicPartition, endOffset);
        emptyConsumer.updateEndOffsets(endOffsetsMap);
        // resetOffsetsTo only seeks the offset, but does not commit.
        streamsResetter.resetOffsetsTo(emptyConsumer, inputTopicPartitions, endOffset + 2L);

        final long position = emptyConsumer.position(topicPartition);

        assertEquals(endOffset, position);
    }