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

[philipp94831]

New KStreamX 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 KStreamX<String, String> input = builder.streamInput();
input.toOutputTopic();

Configuring serdes

Before

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

After

final Serde<SpecificRecord> serde = new SpecificAvroSerde<>();
final KStreamX<String, SpecificRecord> input = builder.streamInput(ConsumedX.valueSerde(serde));
input.toOutputTopic(ProducedX.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, String> deadLetters = processedInput.flatMapValues(ProcessedValue::getErrors)
        .processValues(ErrorHeaderProcessor.withErrorHeaders("Error parsing integers"));
deadLetters.to(builder.getTopics().getErrorTopic());

After

final KStreamX<String, String> input = builder.streamInput();
final KErrorStreamX<String, String, String, Integer> processedInput =
        input.mapValuesCapturingErrors(value -> Integer.parseInt(value));
final KStreamX<String, Integer> intStream = processedInput.values();
intStream.toOutputTopic();
final KStreamX<String, String> deadLetters = processedInput.errors()
        .processValues(ErrorHeaderProcessor.withErrorHeaders("Error parsing integers"));
deadLetters.toErrorTopic();

[jkbe]

I like the new abstractions! In hindsight, I think doing a better job at hiding TopologyBuilder was actually missing from the library the whole time.

[sonarqubecloud]

Quality Gate passed

Issues
6 New issues
427 Accepted issues

Measures
0 Security Hotspots
98.2% Coverage on New Code
0.2% Duplication on New Code

See analysis details on SonarQube Cloud