У меня есть поток потока основной обработки, который выглядит как
master topic -> my processing in a mapper/filter -> output topics
и мне интересно, как лучше всего обращаться с "плохими сообщениями". Возможно, это могут быть такие вещи, как сообщения, которые я не могу десериализовать должным образом, или, возможно, некорректная обработка логики обработки/фильтрации (у меня нет внешних зависимостей, поэтому не должно быть временных ошибок такого рода).
Я рассматривал возможность переноса всего кода обработки/фильтрации в try catch, и если было создано исключение, то маршрутизация в "тему ошибки". Затем я смогу изучить сообщение и изменить его или исправить мой код, а затем воспроизвести его на мастер. Если я допускаю распространение каких-либо исключений, поток, похоже, застрял, и больше сообщений не было получено.
- Этот подход считается наилучшей практикой?
- Есть ли удобный способ управления потоками Kafka? Я не думаю, что есть концепция DLQ...
- Каковы альтернативные способы остановки Kafka jamming при "плохом сообщении"?
- Какие существуют альтернативные подходы к обработке ошибок?
Для полноты здесь мой код (псевдо-иш):
class Document {
// Fields
}
class AnalysedDocument {
Document document;
String rawValue;
Exception exception;
Analysis analysis;
// All being well
AnalysedDocument(Document document, Analysis analysis) {...}
// Analysis failed
AnalysedDocument(Document document, Exception exception) {...}
// Deserialisation failed
AnalysedDocument(String rawValue, Exception exception) {...}
}
KStreamBuilder builder = new KStreamBuilder();
KStream<String, AnalysedPolecatDocument> analysedDocumentStream = builder
.stream(Serdes.String(), Serdes.String(), "master")
.mapValues(new ValueMapper<String, AnalysedDocument>() {
@Override
public AnalysedDocument apply(String rawValue) {
Document document;
try {
// Deserialise
document = ...
} catch (Exception e) {
return new AnalysedDocument(rawValue, exception);
}
try {
// Perform analysis
Analysis analysis = ...
return new AnalysedDocument(document, analysis);
} catch (Exception e) {
return new AnalysedDocument(document, exception);
}
}
});
// Branch based on whether analysis mapping failed to produce errorStream and successStream
errorStream.to(Serdes.String(), customPojoSerde(), "error");
successStream.to(Serdes.String(), customPojoSerde(), "analysed");
KafkaStreams streams = new KafkaStreams(builder, config);
streams.start();
Любая помощь очень ценится.
