JSON Schema Validation and Output Routing

Overview

This article addresses a common issue users may encounter when configuring Redpanda Connect for JSON Schema validation and output routing based on validation results.

Problem Description

Users may experience unexpected behavior when setting up a Redpanda Connect pipeline to:

1. Validate messages from an input topic using JSON Schema
2. Route messages to different output topics based on validation results

The specific issue is that messages are not being routed correctly, even when validation errors occur.

Configuration Example

input:
  kafka_franz:
    # Input configuration details

pipeline:
  processors:
  - json_schema:
      schema: >
        # JSON Schema definition
  - catch:
    - log:
        level: ERROR
        message: "Schema validation failed due to: ${!error()}"
  - mapping: |
      if errored() {
        meta topic = "conduktor_test__v0_failed"
      } else {
        meta topic = "conduktor_test__v0_validated"
      }

output:
  kafka_franz:
    # Output configuration details
    topic: ${! metadata("topic") }

Issue

With the above configuration, messages that fail JSON Schema validation are still being routed to the "validated" topic instead of the "failed" topic.

Solution

The solution involves adjusting the pipeline configuration to ensure proper error handling and routing. Here's an example of a working configuration:

pipeline:
  processors:
  - mapping: |
      meta topic = "connect_test__v0_validated"
  - json_schema:
      schema: >
        # JSON Schema definition
  - catch:
    - log:
        level: ERROR
        message: "Schema validation failed due to: ${!error()}"
    - mapping: |
        meta topic = "connect_test__v0_failed"

This configuration sets a default output topic, then updates it to the "failed" topic if a schema validation error occurs.

Additional Notes

• The catch processor clears out all errors once you leave its scope, which is why the original configuration didn't work as expected.
• It's important to place the error handling and routing logic within the catch block to ensure it's executed when a validation error occurs.

If you need additional support please reach out to Redpanda Support.