class SchemaKStream {

  @SuppressWarnings("unchecked")
  public SchemaKStream selectKey(final Field newKeyField, boolean updateRowKey) {
    if (keyField != null && keyField.name().equals(newKeyField.name())) {
      return this;
    }

    KStream keyedKStream = kstream.filter((key, value) ->
        value != null
            && extractColumn(newKeyField, value) != null
    ).selectKey((key, value) ->
        extractColumn(newKeyField, value)
            .toString()
    ).mapValues((key, row) -> {
      if (updateRowKey) {
        row.getColumns().set(SchemaUtil.ROWKEY_NAME_INDEX, key);
      }
      return row;
    });

    return new SchemaKStream(
        schema,
        keyedKStream,
        newKeyField,
        Collections.singletonList(this),
        Type.REKEY,
        functionRegistry,
        schemaRegistryClient
    );
  }

  public  SchemaKStream(
      final Schema schema,
      final KStream<String, GenericRow> kstream,
      final Field keyField,
      final List<SchemaKStream> sourceSchemaKStreams,
      final Type type,
      final FunctionRegistry functionRegistry,
      final SchemaRegistryClient schemaRegistryClient
  );

}