How did you validate messages from Kafka? What happened if a message was invalid?

For validating messages from Kafka, I usually use schemas (such as Avro, Protobuf, or JSON Schema) and library validators. When receiving a message, I first check its structure and content for compliance with the schema and business rules.

If the message fails validation, different approaches are possible:

• Logging the error and sending the message to a separate error topic (dead-letter queue) to avoid blocking the processing of other messages.
• Ignoring the message with subsequent error monitoring.
• In some cases, attempting to correct the message or request a resend.

Example in Go using JSON Schema for validation:

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
}

Thus, in case of an invalid message, it can be sent to a dead-letter topic for subsequent analysis.