class KsqlStructuredDataOutputNode extends OutputNode {

  @Override
  public SchemaKStream buildStream(
      final StreamsBuilder builder,
      final KsqlConfig ksqlConfig,
      final KafkaTopicClient kafkaTopicClient,
      final FunctionRegistry functionRegistry,
      final Map<String, Object> props,
      final SchemaRegistryClient schemaRegistryClient
  ) {
    final Map<String, Object> outputProperties = getOutputProperties();
    final PlanNode source = getSource();
    final SchemaKStream schemaKStream = source.buildStream(
        builder,
        ksqlConfig,
        kafkaTopicClient,
        functionRegistry,
        props,
        schemaRegistryClient
    );

    final Set<Integer> rowkeyIndexes = SchemaUtil.getRowTimeRowKeyIndexes(getSchema());
    final Builder outputNodeBuilder = Builder.from(this);
    final Schema schema = SchemaUtil.removeImplicitRowTimeRowKeyFromSchema(getSchema());
    outputNodeBuilder.withSchema(schema);

    if (getTopicSerde() instanceof KsqlAvroTopicSerDe) {
      addAvroSchemaToResultTopic(outputNodeBuilder);
    }


    if (outputProperties.containsKey(KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY)) {
      ksqlConfig.put(
          KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY,
          outputProperties.get(KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY)
      );
    }
    if (outputProperties.containsKey(KsqlConfig.SINK_NUMBER_OF_REPLICAS_PROPERTY)) {
      ksqlConfig.put(
          KsqlConfig.SINK_NUMBER_OF_REPLICAS_PROPERTY,
          outputProperties.get(KsqlConfig.SINK_NUMBER_OF_REPLICAS_PROPERTY)
      );
    }

    final SchemaKStream result = createOutputStream(
        schemaKStream,
        outputNodeBuilder,
        functionRegistry,
        outputProperties,
        schemaRegistryClient
    );

    final KsqlStructuredDataOutputNode noRowKey = outputNodeBuilder.build();
    createSinkTopic(noRowKey.getKafkaTopicName(),
                    ksqlConfig,
                    kafkaTopicClient,
                    shouldBeCompacted(result));
    result.into(
        noRowKey.getKafkaTopicName(),
        noRowKey.getKsqlTopic().getKsqlTopicSerDe()
            .getGenericRowSerde(noRowKey.getSchema(), ksqlConfig, false, schemaRegistryClient),
        rowkeyIndexes
    );

    result.setOutputNode(
        outputNodeBuilder
            .withSchema(
                SchemaUtil.addImplicitRowTimeRowKeyToSchema(noRowKey.getSchema()))
            .build());
    return result;
  }

  @JsonCreator public  KsqlStructuredDataOutputNode(
      @JsonProperty("id") final PlanNodeId id,
      @JsonProperty("source") final PlanNode source,
      @JsonProperty("schema") final Schema schema,
      @JsonProperty("timestamp") final TimestampExtractionPolicy timestampExtractionPolicy,
      @JsonProperty("key") final Field keyField,
      @JsonProperty("ksqlTopic") final KsqlTopic ksqlTopic,
      @JsonProperty("topicName") final String topicName,
      @JsonProperty("outputProperties") final Map<String, Object> outputProperties,
      @JsonProperty("limit") final Optional<Integer> limit);

}