class SchemaKStream {

  @SuppressWarnings("unchecked")
  public SchemaKStream filter(final Expression filterExpression) {
    SqlPredicate predicate = new SqlPredicate(filterExpression, schema, false, functionRegistry);
    KStream<String, GenericRow> filteredKStream = kstream.filter(predicate.getPredicate());
    return new SchemaKStream(
        schema,
        filteredKStream,
        keyField,
        Arrays.asList(this),
        Type.FILTER,
        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
  );

}