public static RecordBatchingStateRestoreCallback adapt(final StateRestoreCallback restoreCallback) {
        Objects.requireNonNull(restoreCallback, "stateRestoreCallback must not be null");
        if (restoreCallback instanceof RecordBatchingStateRestoreCallback) {
            return (RecordBatchingStateRestoreCallback) restoreCallback;
        } else if (restoreCallback instanceof BatchingStateRestoreCallback) {
            return records -> {
                final List<KeyValue<byte[], byte[]>> keyValues = new ArrayList<>();
                for (final ConsumerRecord<byte[], byte[]> record : records) {
                    keyValues.add(new KeyValue<>(record.key(), record.value()));
                }
                ((BatchingStateRestoreCallback) restoreCallback).restoreAll(keyValues);
            };
        } else {
            return records -> {
                for (final ConsumerRecord<byte[], byte[]> record : records) {
                    restoreCallback.restore(record.key(), record.value());
                }
            };
        }
    }
@Test
    public void shouldConvertToKeyValueBatches() {
        final ArrayList<KeyValue<byte[], byte[]>> actual = new ArrayList<>();
        final BatchingStateRestoreCallback callback = new BatchingStateRestoreCallback() {
            @Override
            public void restoreAll(final Collection<KeyValue<byte[], byte[]>> records) {
                actual.addAll(records);
            }

            @Override
            public void restore(final byte[] key, final byte[] value) {
                // unreachable
            }
        };

        final RecordBatchingStateRestoreCallback adapted = adapt(callback);

        final byte[] key1 = {1};
        final byte[] value1 = {2};
        final byte[] key2 = {3};
        final byte[] value2 = {4};
        adapted.restoreBatch(asList(
            new ConsumerRecord<>("topic1", 0, 0L, key1, value1),
            new ConsumerRecord<>("topic2", 1, 1L, key2, value2)
        ));

        assertThat(
            actual,
            is(asList(
                new KeyValue<>(key1, value1),
                new KeyValue<>(key2, value2)
            ))
        );
    }