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

    schemas.trackKeySerdeCreation(
        loggerNamePrefix,
        schema.logicalSchema(),
        KeyFormat.nonWindowed(format, schema.keySchema().features())
    );

    return keySerdeFactory.create(
        format,
        schema.keySchema(),
        ksqlConfig,
        serviceContext.getSchemaRegistryClientFactory(),
        loggerNamePrefix,
        processingLogContext,
        getSerdeTracker(loggerNamePrefix)
    );
  }
@Test
  public void shouldBuildWindowedKeySerde() {
    // Then:
    runtimeBuildContext.buildKeySerde(
        FORMAT_INFO,
        WINDOW_INFO,
        PHYSICAL_SCHEMA,
        queryContext
    );

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