public Serde<GenericRow> buildValueSerde(
      final FormatInfo format,
      final PhysicalSchema schema,
      final QueryContext queryContext
  ) {
    final String loggerNamePrefix = QueryLoggerUtil.queryLoggerName(queryId, queryContext);

    schemas.trackValueSerdeCreation(
        loggerNamePrefix,
        schema.logicalSchema(),
        ValueFormat.of(format, schema.valueSchema().features())
    );

    return valueSerdeFactory.create(
        format,
        schema.valueSchema(),
        ksqlConfig,
        serviceContext.getSchemaRegistryClientFactory(),
        loggerNamePrefix,
        processingLogContext,
        getSerdeTracker(loggerNamePrefix)
    );
  }
@Test
  public void shouldBuildValueSerde() {
    // Then:
    runtimeBuildContext.buildValueSerde(
        FORMAT_INFO,
        PHYSICAL_SCHEMA,
        queryContext
    );

    // Then:
    verify(valueSerdeFactory).create(
        FORMAT_INFO,
        PHYSICAL_SCHEMA.valueSchema(),
        ksqlConfig,
        srClientFactory,
        QueryLoggerUtil.queryLoggerName(QUERY_ID, queryContext),
        processingLogContext,
        Optional.empty()
    );
  }