@Override
  public Serde<GenericRow> create(
      final FormatInfo format,
      final PersistenceSchema schema,
      final KsqlConfig ksqlConfig,
      final Supplier<SchemaRegistryClient> srClientFactory,
      final String loggerNamePrefix,
      final ProcessingLogContext processingLogContext,
      final Optional<TrackedCallback> tracker
  ) {
    final Serde<List<?>> formatSerde =
        innerFactory.createFormatSerde("Value", format, schema, ksqlConfig, srClientFactory, false);

    final Serde<GenericRow> genericRowSerde = toGenericRowSerde(formatSerde, schema);

    final Serde<GenericRow> loggingSerde = innerFactory.wrapInLoggingSerde(
        genericRowSerde,
        loggerNamePrefix,
        processingLogContext,
        queryId);

    final Serde<GenericRow> serde = tracker
        .map(callback -> innerFactory.wrapInTrackingSerde(loggingSerde, callback))
        .orElse(loggingSerde);

    serde.configure(Collections.emptyMap(), false);

    return serde;
  }
@Test
  public void shouldWrapInGenericSerde() {
    // When:
    factory.create(format, schema, config, srClientFactory, LOGGER_PREFIX, processingLogCxt,
        Optional.empty());

    // Then:
    verify(innerFactory).wrapInLoggingSerde(rowSerdeCaptor.capture(), any(), any(), any());

    assertThat(rowSerdeCaptor.getValue().serializer(), is(instanceOf(GenericSerializer.class)));
    assertThat(rowSerdeCaptor.getValue().deserializer(),
        is(instanceOf(GenericDeserializer.class)));
  }