private void resetConsumerOffset(String topic, int partiontionNum, long offset) {
        Consumer consumer = testConsumerFactory.createConsumer();
        TopicPartition partition = new TopicPartition(topic, partiontionNum);
        consumer.assign(Arrays.asList(partition));
        consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(offset + 1)));
    }

    private void resetConsumerOffsetToEnd(String topic, int partiontionNum) {
        Consumer consumer = testConsumerFactory.createConsumer();
        TopicPartition partition = new TopicPartition(topic, partiontionNum);
        consumer.assign(Arrays.asList(partition));
        consumer.seekToEnd();
        long position = consumer.position(partition);
        consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(position)));
    }