Как вы делали валидацию сообщений из Kafka? Если сообщение не провалидировалось — что с ним происходило?

Для валидации сообщений из Kafka обычно использую схемы (например, Avro, Protobuf или JSON Schema) и библиотечные валидаторы. При получении сообщения сначала проверяю его структуру и содержимое на соответствие схеме и бизнес-правилам.

Если сообщение не проходит валидацию, возможны разные подходы:

• Логирование ошибки и отправка сообщения в отдельный топик для ошибок (dead-letter queue), чтобы не блокировать обработку остальных сообщений.
• Игнорирование сообщения с последующим мониторингом ошибок.
• В некоторых случаях — попытка исправить сообщение или запросить повторную отправку.

Пример на Go с использованием JSON Schema для валидации:

import (
    "github.com/xeipuuv/gojsonschema"
)

func validateMessage(jsonData []byte, schemaLoader gojsonschema.JSONLoader) (bool, error) {
    documentLoader := gojsonschema.NewBytesLoader(jsonData)
    result, err := gojsonschema.Validate(schemaLoader, documentLoader)
    if err != nil {
        return false, err
    }
    return result.Valid(), nil
}

Таким образом, при невалидном сообщении можно отправить его в dead-letter топик для последующего анализа.