class StateRestorer {

    void restore(final Collection<ConsumerRecord<byte[], byte[]>> records) {
        final Collection<ConsumerRecord<byte[], byte[]>> convertedRecords = new ArrayList<>(records.size());
        for (final ConsumerRecord<byte[], byte[]> record : records) {
            convertedRecords.add(recordConverter.convert(record));
        }
        compositeRestoreListener.restoreBatch(convertedRecords);
    }

      StateRestorer(final TopicPartition partition,
                  final CompositeRestoreListener compositeRestoreListener,
                  final Long checkpoint,
                  final long offsetLimit,
                  final boolean persistent,
                  final String storeName,
                  final RecordConverter recordConverter);

}