public static GenericSchemaImpl of(SchemaInfo schemaInfo) {
        return of(schemaInfo, true);
    }
@Test
    public void testKeyValueSchema() {
        // configure the schema info provider
        MultiVersionSchemaInfoProvider multiVersionSchemaInfoProvider = mock(MultiVersionSchemaInfoProvider.class);

        List<Schema<Foo>> encodeSchemas = Lists.newArrayList(
            Schema.JSON(Foo.class),
            Schema.AVRO(Foo.class)
        );

        for (Schema<Foo> keySchema : encodeSchemas) {
            for (Schema<Foo> valueSchema : encodeSchemas) {
                // configure encode schema
                Schema<KeyValue<Foo, Foo>> kvSchema = KeyValueSchemaImpl.of(
                    keySchema, valueSchema
                );

                // configure decode schema
                Schema<KeyValue<GenericRecord, GenericRecord>> decodeSchema = KeyValueSchemaImpl.of(
                    Schema.AUTO_CONSUME(), Schema.AUTO_CONSUME()
                );
                decodeSchema.configureSchemaInfo(
                    "test-topic", "topic",kvSchema.getSchemaInfo()
                );

                GenericSchema genericAvroSchema = GenericSchemaImpl.of(Schema.AVRO(Foo.class).getSchemaInfo());
                when(multiVersionSchemaInfoProvider.getSchemaByVersion(any(byte[].class)))
                        .thenReturn(CompletableFuture.completedFuture(
                                KeyValueSchemaInfo.encodeKeyValueSchemaInfo(
                                        keySchema,
                                        valueSchema,
                                        KeyValueEncodingType.INLINE
                                )
                        ));
                decodeSchema.setSchemaInfoProvider(multiVersionSchemaInfoProvider);

                testEncodeAndDecodeKeyValues(kvSchema, decodeSchema);
            }
        }

    }