Use custom streams implementation to simplify error handling, writing to topics and serde configuration

[philipp94831]

New ImproveKStream interface adds new methods to simplify topology creation. It inherits from KStream and therefore no changes to existing code are required.

Writing to topics

Before

final KStream<String, String> input = builder.streamInput();
input.to(builder.getTopics().getOutputTopic());

After

final ImprovedKStream<String, String> input = builder.streamInput();
input.toOutputTopic();

Configuring serdes

Before

final Preconfigured<Serde<SpecificRecord>> configuredSerde = Preconfigured.create(new SpecificAvroSerde<>());
final Serde<SpecificRecord> serde = builder.createConfigurator().configureForValues(configuredSerde);
final KStream<String, SpecificRecord> input = builder.streamInput(Consumed.with(null, serde));
input.to(builder.getTopics().getOutputTopic(), Produced.valueSerde(serde));

After

final Preconfigured<Serde<SpecificRecord>> serde = Preconfigured.create(new SpecificAvroSerde<>());
final ImprovedKStream<String, SpecificRecord> input = builder.streamInput(ConfiguredConsumed.valueSerde(serde));
input.toOutputTopic(ConfiguredProduced.valueSerde(serde));

Error handling

Before

final KStream<String, String> input = builder.streamInput();
final KStream<String, ProcessedValue<String, Integer>> processedInput =
        input.mapValues(ErrorCapturingValueMapper.captureErrors(value -> Integer.parseInt(value)));
final KStream<String, Integer> intStream = processedInput.flatMapValues(ProcessedValue::getValues);
intStream.to(builder.getTopics().getOutputTopic());
final KStream<String, DeadLetter> deadLetters = processedInput.flatMapValues(ProcessedValue::getErrors)
        .processValues(AvroDeadLetterConverter.asProcessor("Error parsing integers"));
deadLetters.to(builder.getTopics().getErrorTopic());

After

final ImprovedKStream<String, String> input = builder.streamInput();
final KErrorStream<String, String, String, Integer> processedInput =
        input.mapValuesCapturingErrors(value -> Integer.parseInt(value));
final ImprovedKStream<String, Integer> intStream = processedInput.values();
intStream.toOutputTopic();
final ImprovedKStream<String, DeadLetter> deadLetters = processedInput.errors()
        .processValues(AvroDeadLetterConverter.asProcessor("Error parsing integers"));
deadLetters.toErrorTopic();