Skip to main content
Quick navigation

Schema Id validation

View the full demo in realtime​

You can either follow all the steps manually, or watch the recording

Review the docker compose environment​

As can be seen from docker-compose.yaml the demo environment consists of the following services:

  • gateway1
  • gateway2
  • kafka-client
  • kafka1
  • kafka2
  • kafka3
  • schema-registry
cat docker-compose.yaml

Starting the docker environment​

Start all your docker processes, wait for them to be up and ready, then run in background

  • --wait: Wait for services to be running|healthy. Implies detached mode.
  • --detach: Detached mode: Run containers in the background
docker compose up --detach --wait

Creating topic users on gateway1​

Creating on gateway1:

  • Topic users with partitions:1 and replication-factor:1
kafka-topics \
    --bootstrap-server localhost:6969 \
    --replication-factor 1 \
    --partitions 1 \
    --create --if-not-exists \
    --topic users

Listing topics in gateway1​

kafka-topics \
    --bootstrap-server localhost:6969 \
    --list

Adding interceptor schema-id​

step-07-schema-id-interceptor.json:

{
  "kind" : "Interceptor",
  "apiVersion" : "gateway/v2",
  "metadata" : {
    "name" : "schema-id"
  },
  "spec" : {
    "comment" : "Adding interceptor: schema-id",
    "pluginClass" : "io.conduktor.gateway.interceptor.safeguard.TopicRequiredSchemaIdPolicyPlugin",
    "priority" : 100,
    "config" : {
      "topic" : "users",
      "schemaIdRequired" : true
    }
  }
}
curl \
    --silent \
    --request PUT "http://localhost:8888/gateway/v2/interceptor" \
    --header "Content-Type: application/json" \
    --user "admin:conduktor" \
    --data @step-07-schema-id-interceptor.json | jq

Listing interceptors​

Listing interceptors on gateway1

curl \
    --silent \
    --request GET "http://localhost:8888/gateway/v2/interceptor" \
    --user "admin:conduktor" | jq

Producing 1 message in users​

Producing 1 message in users in cluster gateway1

Sending 1 event

{
  "msg" : "hello world"
}
echo '{"msg":"hello world"}' | \
    kafka-console-producer \
        --bootstrap-server localhost:6969 \
        --topic users

[!IMPORTANT] We get the following exception

org.apache.kafka.common.errors.PolicyViolationException:
> Request parameters do not satisfy the configured policy. Request parameters do not satisfy the configured policy.
>Topic 'users' with schemaId is required.

Consuming from users​

Consuming from users in cluster gateway1

kafka-console-consumer \
    --bootstrap-server localhost:6969 \
    --topic users \
    --from-beginning \
    --max-messages 1 \
    --timeout-ms 3000 | jq

Send avro message​

echo '{
    "name": "conduktor",
    "username": "test@conduktor.io",
    "password": "password1",
    "visa": "visa123456",
    "address": "Conduktor Towers, London"
}' | \
  jq -c | \
      kafka-json-schema-console-producer  \
        --bootstrap-server localhost:6969 \
        --topic users \
        --property schema.registry.url=http://localhost:8081 \
        --property value.schema='{
            "title": "User",
            "type": "object",
            "properties": {
                "name": { "type": "string" },
                "username": { "type": "string" },
                "password": { "type": "string" },
                "visa": { "type": "string" },
                "address": { "type": "string" }
            }
        }'

Get subjects​

curl --silent http://localhost:8081/subjects/ | jq

Consuming from users​

Consuming from users in cluster gateway1

kafka-console-consumer \
    --bootstrap-server localhost:6969 \
    --topic users \
    --from-beginning \
    --max-messages 2 \
    --timeout-ms 3000 | jq

Tearing down the docker environment​

Remove all your docker processes and associated volumes

  • --volumes: Remove named volumes declared in the "volumes" section of the Compose file and anonymous volumes attached to containers.
docker compose down --volumes

Conclusion

You can now make sure you don't fall because of a wrong message