What is a Schema Payload Validation Policy Interceptor?
Avoid outages from missing or badly formatted records, ensure all messages adhere to a schema.
This interceptor also supports validating payload against specific constraints for AvroSchema and JsonSchema.
This is similar to the validations provided by JsonSchema, such as:
- Number:
minimum
,maximum
,exclusiveMinimum
,exclusiveMaximum
,multipleOf
- String:
minLength
,maxLength
,pattern
,format
- Collections:
maxItems
,minItems
This interceptor also supports validating payload against specific custom constraints expression
,
which uses a simple language familiar with devs is CEL (Common Expression Language)
This interceptor also supports validating payload against specific custom metadata.rules
object in the schema
using CEL, too.
View the full demo in realtime
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
- zookeeper
- Command
- File Content
cat docker-compose.yaml
version: '3.7'
services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
hostname: zookeeper
container_name: zookeeper
environment:
ZOOKEEPER_CLIENT_PORT: 2801
ZOOKEEPER_TICK_TIME: 2000
healthcheck:
test: nc -zv 0.0.0.0 2801 || exit 1
interval: 5s
retries: 25
kafka1:
hostname: kafka1
container_name: kafka1
image: confluentinc/cp-kafka:latest
ports:
- 19092:19092
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2801
KAFKA_LISTENERS: INTERNAL://:9092,EXTERNAL_SAME_HOST://:19092
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:9092,EXTERNAL_SAME_HOST://localhost:19092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL_SAME_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_LOG4J_LOGGERS: kafka.authorizer.logger=INFO
KAFKA_LOG4J_ROOT_LOGLEVEL: WARN
KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
depends_on:
zookeeper:
condition: service_healthy
healthcheck:
test: nc -zv kafka1 9092 || exit 1
interval: 5s
retries: 25
kafka2:
hostname: kafka2
container_name: kafka2
image: confluentinc/cp-kafka:latest
ports:
- 19093:19093
environment:
KAFKA_BROKER_ID: 2
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2801
KAFKA_LISTENERS: INTERNAL://:9093,EXTERNAL_SAME_HOST://:19093
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:9093,EXTERNAL_SAME_HOST://localhost:19093
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL_SAME_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_LOG4J_LOGGERS: kafka.authorizer.logger=INFO
KAFKA_LOG4J_ROOT_LOGLEVEL: WARN
KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
depends_on:
zookeeper:
condition: service_healthy
healthcheck:
test: nc -zv kafka2 9093 || exit 1
interval: 5s
retries: 25
kafka3:
image: confluentinc/cp-kafka:latest
hostname: kafka3
container_name: kafka3
ports:
- 19094:19094
environment:
KAFKA_BROKER_ID: 3
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2801
KAFKA_LISTENERS: INTERNAL://:9094,EXTERNAL_SAME_HOST://:19094
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:9094,EXTERNAL_SAME_HOST://localhost:19094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL_SAME_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_LOG4J_LOGGERS: kafka.authorizer.logger=INFO
KAFKA_LOG4J_ROOT_LOGLEVEL: WARN
KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
depends_on:
zookeeper:
condition: service_healthy
healthcheck:
test: nc -zv kafka3 9094 || exit 1
interval: 5s
retries: 25
schema-registry:
image: confluentinc/cp-schema-registry:latest
hostname: schema-registry
container_name: schema-registry
ports:
- 8081:8081
environment:
SCHEMA_REGISTRY_HOST_NAME: schema-registry
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka1:9092,kafka2:9093,kafka3:9094
SCHEMA_REGISTRY_LOG4J_ROOT_LOGLEVEL: WARN
SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
SCHEMA_REGISTRY_SCHEMA_REGISTRY_GROUP_ID: schema-registry
volumes:
- type: bind
source: .
target: /clientConfig
read_only: true
depends_on:
kafka1:
condition: service_healthy
kafka2:
condition: service_healthy
kafka3:
condition: service_healthy
healthcheck:
test: nc -zv schema-registry 8081 || exit 1
interval: 5s
retries: 25
gateway1:
image: conduktor/conduktor-gateway:3.0.3
hostname: gateway1
container_name: gateway1
environment:
KAFKA_BOOTSTRAP_SERVERS: kafka1:9092,kafka2:9093,kafka3:9094
GATEWAY_ADVERTISED_HOST: localhost
GATEWAY_MODE: VCLUSTER
GATEWAY_SECURITY_PROTOCOL: SASL_PLAINTEXT
GATEWAY_FEATURE_FLAGS_ANALYTICS: false
depends_on:
kafka1:
condition: service_healthy
kafka2:
condition: service_healthy
kafka3:
condition: service_healthy
ports:
- 6969:6969
- 6970:6970
- 6971:6971
- 8888:8888
healthcheck:
test: curl localhost:8888/health
interval: 5s
retries: 25
gateway2:
image: conduktor/conduktor-gateway:3.0.3
hostname: gateway2
container_name: gateway2
environment:
KAFKA_BOOTSTRAP_SERVERS: kafka1:9092,kafka2:9093,kafka3:9094
GATEWAY_ADVERTISED_HOST: localhost
GATEWAY_MODE: VCLUSTER
GATEWAY_SECURITY_PROTOCOL: SASL_PLAINTEXT
GATEWAY_FEATURE_FLAGS_ANALYTICS: false
GATEWAY_START_PORT: 7969
depends_on:
kafka1:
condition: service_healthy
kafka2:
condition: service_healthy
kafka3:
condition: service_healthy
ports:
- 7969:7969
- 7970:7970
- 7971:7971
- 8889:8888
healthcheck:
test: curl localhost:8888/health
interval: 5s
retries: 25
kafka-client:
image: confluentinc/cp-kafka:latest
hostname: kafka-client
container_name: kafka-client
command: sleep infinity
volumes:
- type: bind
source: .
target: /clientConfig
read_only: true
networks:
demo: null
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 berunning|healthy
. Implies detached mode.--detach
: Detached mode: Run containers in the background
- Command
- Output
- Recording
docker compose up --detach --wait
Network safeguard-validate-schema-payload-proto_default Creating
Network safeguard-validate-schema-payload-proto_default Created
Container kafka-client Creating
Container zookeeper Creating
Container kafka-client Created
Container zookeeper Created
Container kafka1 Creating
Container kafka3 Creating
Container kafka2 Creating
Container kafka1 Created
Container kafka3 Created
Container kafka2 Created
Container gateway2 Creating
Container schema-registry Creating
Container gateway1 Creating
Container gateway2 Created
Container gateway1 Created
Container schema-registry Created
Container kafka-client Starting
Container zookeeper Starting
Container kafka-client Started
Container zookeeper Started
Container zookeeper Waiting
Container zookeeper Waiting
Container zookeeper Waiting
Container zookeeper Healthy
Container kafka1 Starting
Container zookeeper Healthy
Container kafka3 Starting
Container zookeeper Healthy
Container kafka2 Starting
Container kafka1 Started
Container kafka2 Started
Container kafka3 Started
Container kafka2 Waiting
Container kafka3 Waiting
Container kafka1 Waiting
Container kafka3 Waiting
Container kafka1 Waiting
Container kafka2 Waiting
Container kafka3 Waiting
Container kafka1 Waiting
Container kafka2 Waiting
Container kafka2 Healthy
Container kafka1 Healthy
Container kafka3 Healthy
Container kafka1 Healthy
Container kafka2 Healthy
Container kafka3 Healthy
Container kafka1 Healthy
Container gateway2 Starting
Container kafka2 Healthy
Container schema-registry Starting
Container kafka3 Healthy
Container gateway1 Starting
Container gateway1 Started
Container gateway2 Started
Container schema-registry Started
Container zookeeper Waiting
Container kafka1 Waiting
Container kafka2 Waiting
Container kafka3 Waiting
Container schema-registry Waiting
Container gateway1 Waiting
Container gateway2 Waiting
Container kafka-client Waiting
Container kafka3 Healthy
Container kafka1 Healthy
Container kafka-client Healthy
Container kafka2 Healthy
Container zookeeper Healthy
Container gateway1 Healthy
Container schema-registry Healthy
Container gateway2 Healthy
Creating virtual cluster teamA
Creating virtual cluster teamA
on gateway gateway1
and reviewing the configuration file to access it
- Command
- Output
- Recording
# Generate virtual cluster teamA with service account sa
token=$(curl \
--request POST "http://localhost:8888/admin/vclusters/v1/vcluster/teamA/username/sa" \
--header 'Content-Type: application/json' \
--user 'admin:conduktor' \
--silent \
--data-raw '{"lifeTimeSeconds": 7776000}' | jq -r ".token")
# Create access file
echo """
bootstrap.servers=localhost:6969
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='sa' password='$token';
""" > teamA-sa.properties
# Review file
cat teamA-sa.properties
bootstrap.servers=localhost:6969
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='sa' password='eyJhbGciOiJIUzI1NiJ9.eyJ1c2VybmFtZSI6InNhIiwidmNsdXN0ZXIiOiJ0ZWFtQSIsImV4cCI6MTcyMDQ4MzExMH0.3OYpRHHjWQPiAyE1C7mqbNWpNJ3SjTPUOWx3VCE6fFA';
Creating topic topic-protobuf on teamA
Creating on teamA
:
- Topic
topic-protobuf
with partitions:1 and replication-factor:1
- Command
- Output
- Recording
Review the example protocol buffer schema
Review the example protocol buffer schema
- Command
- File Content
cat user-schema.proto
syntax = "proto3";
option java_package = "schema.protobuf";
option java_outer_classname = "User";
message Student {
string name = 1;
int32 age = 2;
string email = 3;
Address address = 4;
repeated string hobbies = 5;
repeated Friend friends = 6;
message Address {
string street = 1;
string city = 2;
}
message Friend {
string name = 1;
int32 age = 2;
}
}
Let's register it to the Schema Registry
- Command
- Output
- Recording
Review invalid payload
Review invalid payload
- Command
- File Content
cat invalid-payload.json
{
"name": "D",
"age": 17,
"email": "bad email",
"address": {
"street": "a way too lond adress that will not fit in your database",
"city": ""
},
"hobbies": [
"reading"
],
"friends": [
{
"name": "Tommy",
"age": 17
},
{
"name": "Emma Watson",
"age": 18
}
]
}
Let's send invalid data
- Command
- Output
- Recording
cat invalid-payload.json | jq -c | \
kafka-protobuf-console-producer \
--bootstrap-server localhost:6969 \
--producer.config teamA-sa.properties \
--topic topic-protobuf \
--property schema.registry.url=http://localhost:8081 \
--property value.schema.id=1
[2024-04-10 03:58:32,915] INFO KafkaProtobufSerializerConfig values:
auto.register.schemas = true
basic.auth.credentials.source = URL
basic.auth.user.info = [hidden]
bearer.auth.cache.expiry.buffer.seconds = 300
bearer.auth.client.id = null
bearer.auth.client.secret = null
bearer.auth.credentials.source = STATIC_TOKEN
bearer.auth.custom.provider.class = null
bearer.auth.identity.pool.id = null
bearer.auth.issuer.endpoint.url = null
bearer.auth.logical.cluster = null
bearer.auth.scope = null
bearer.auth.scope.claim.name = scope
bearer.auth.sub.claim.name = sub
bearer.auth.token = [hidden]
context.name.strategy = class io.confluent.kafka.serializers.context.NullContextNameStrategy
http.connect.timeout.ms = 60000
http.read.timeout.ms = 60000
id.compatibility.strict = true
key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
latest.cache.size = 1000
latest.cache.ttl.sec = -1
latest.compatibility.strict = true
max.schemas.per.subject = 1000
normalize.schemas = false
proxy.host =
proxy.port = -1
reference.lookup.only = false
reference.subject.name.strategy = class io.confluent.kafka.serializers.subject.DefaultReferenceSubjectNameStrategy
rule.actions = []
rule.executors = []
rule.service.loader.enable = true
schema.format = null
schema.reflection = false
schema.registry.basic.auth.user.info = [hidden]
schema.registry.ssl.cipher.suites = null
schema.registry.ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
schema.registry.ssl.endpoint.identification.algorithm = https
schema.registry.ssl.engine.factory.class = null
schema.registry.ssl.key.password = null
schema.registry.ssl.keymanager.algorithm = SunX509
schema.registry.ssl.keystore.certificate.chain = null
schema.registry.ssl.keystore.key = null
schema.registry.ssl.keystore.location = null
schema.registry.ssl.keystore.password = null
schema.registry.ssl.keystore.type = JKS
schema.registry.ssl.protocol = TLSv1.3
schema.registry.ssl.provider = null
schema.registry.ssl.secure.random.implementation = null
schema.registry.ssl.trustmanager.algorithm = PKIX
schema.registry.ssl.truststore.certificates = null
schema.registry.ssl.truststore.location = null
schema.registry.ssl.truststore.password = null
schema.registry.ssl.truststore.type = JKS
schema.registry.url = [http://localhost:8081]
skip.known.types = true
use.latest.version = false
use.latest.with.metadata = null
use.schema.id = -1
value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
(io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializerConfig:376)
Let's consume it back
That's pretty bad, you are going to propagate wrong data within your system!
- Command
- Output
- Recording
kafka-protobuf-console-consumer \
--bootstrap-server localhost:6969 \
--consumer.config teamA-sa.properties \
--topic topic-protobuf \
--from-beginning \
--timeout-ms 3000
[2024-04-10 03:58:34,463] INFO KafkaProtobufDeserializerConfig values:
auto.register.schemas = true
basic.auth.credentials.source = URL
basic.auth.user.info = [hidden]
bearer.auth.cache.expiry.buffer.seconds = 300
bearer.auth.client.id = null
bearer.auth.client.secret = null
bearer.auth.credentials.source = STATIC_TOKEN
bearer.auth.custom.provider.class = null
bearer.auth.identity.pool.id = null
bearer.auth.issuer.endpoint.url = null
bearer.auth.logical.cluster = null
bearer.auth.scope = null
bearer.auth.scope.claim.name = scope
bearer.auth.sub.claim.name = sub
bearer.auth.token = [hidden]
context.name.strategy = class io.confluent.kafka.serializers.context.NullContextNameStrategy
derive.type = false
http.connect.timeout.ms = 60000
http.read.timeout.ms = 60000
id.compatibility.strict = true
key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
latest.cache.size = 1000
latest.cache.ttl.sec = -1
latest.compatibility.strict = true
max.schemas.per.subject = 1000
normalize.schemas = false
proxy.host =
proxy.port = -1
rule.actions = []
rule.executors = []
rule.service.loader.enable = true
schema.format = null
schema.reflection = false
schema.registry.basic.auth.user.info = [hidden]
schema.registry.ssl.cipher.suites = null
schema.registry.ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
schema.registry.ssl.endpoint.identification.algorithm = https
schema.registry.ssl.engine.factory.class = null
schema.registry.ssl.key.password = null
schema.registry.ssl.keymanager.algorithm = SunX509
schema.registry.ssl.keystore.certificate.chain = null
schema.registry.ssl.keystore.key = null
schema.registry.ssl.keystore.location = null
schema.registry.ssl.keystore.password = null
schema.registry.ssl.keystore.type = JKS
schema.registry.ssl.protocol = TLSv1.3
schema.registry.ssl.provider = null
schema.registry.ssl.secure.random.implementation = null
schema.registry.ssl.trustmanager.algorithm = PKIX
schema.registry.ssl.truststore.certificates = null
schema.registry.ssl.truststore.location = null
schema.registry.ssl.truststore.password = null
schema.registry.ssl.truststore.type = JKS
schema.registry.url = [http://localhost:8081]
specific.protobuf.key.type = class java.lang.Object
specific.protobuf.value.type = class java.lang.Object
use.latest.version = false
use.latest.with.metadata = null
use.schema.id = -1
value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
(io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializerConfig:376)
{"name":"D","age":17,"email":"bad email","address":{"street":"a way too lond adress that will not fit in your database","city":""},"hobbies":["reading"],"friends":[{"name":"Tommy","age":17},{"name":"Emma Watson","age":18}]}
[2024-04-10 03:58:38,194] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$:44)
org.apache.kafka.common.errors.TimeoutException
[2024-04-10 03:58:38,194] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$:44)
org.apache.kafka.common.errors.TimeoutException
Processed a total of 1 messages
Adding interceptor guard-schema-payload-validate
Add Schema Payload Validation Policy Interceptor
- Command
- Output
- Recording
cat step-12-guard-schema-payload-validate.json | jq
curl \
--request POST "http://localhost:8888/admin/interceptors/v1/vcluster/teamA/interceptor/guard-schema-payload-validate" \
--header 'Content-Type: application/json' \
--user 'admin:conduktor' \
--silent \
--data @step-12-guard-schema-payload-validate.json | jq
{
"pluginClass": "io.conduktor.gateway.interceptor.safeguard.SchemaPayloadValidationPolicyPlugin",
"priority": 100,
"config": {
"schemaRegistryConfig": {
"host": "http://schema-registry:8081"
},
"topic": "topic-.*",
"schemaIdRequired": true,
"validateSchema": true,
"action": "BLOCK"
}
}
{
"message": "guard-schema-payload-validate is created"
}
Listing interceptors for teamA
Listing interceptors on gateway1
for virtual cluster teamA
- Command
- Output
- Recording
curl \
--request GET 'http://localhost:8888/admin/interceptors/v1/vcluster/teamA' \
--header 'Content-Type: application/json' \
--user 'admin:conduktor' \
--silent | jq
{
"interceptors": [
{
"name": "guard-schema-payload-validate",
"pluginClass": "io.conduktor.gateway.interceptor.safeguard.SchemaPayloadValidationPolicyPlugin",
"priority": 100,
"timeoutMs": 9223372036854775807,
"config": {
"schemaRegistryConfig": {
"host": "http://schema-registry:8081"
},
"topic": "topic-.*",
"schemaIdRequired": true,
"validateSchema": true,
"action": "BLOCK"
}
}
]
}
Review the protocol buffer schema with validation rules
Review the protocol buffer schema with validation rules
- Command
- File Content
cat user-schema-with-validation-rules.proto
syntax = "proto3";
option java_package = "schema.protobuf";
option java_outer_classname = "User";
message Student {
option (confluent.message_meta).params = {
metadata: "{\"rules\":[{\"name\":\"check name\",\"expression\":\"size(message.name) > 2\",\"message\":\"name length must greater than 2\"},{\"name\":\"checkAge\",\"expression\":\"message.age >= 18\",\"message\":\"age must be greater than or equal to 18\"}]}"
};
string name = 1 [(confluent.field_meta).params = {minLength: "3", maxLength: "50", expression: "size(name) >= 3 && size(name) <= 50"}];
int32 age = 2 [(confluent.field_meta).params = {minimum: "3", maximum: "120", expression: "age >= 3 && age <= 120"}];
string email = 3 [(confluent.field_meta).params = {format: "email", expression: "email.contains('foo')"}];
Address address = 4;
repeated string hobbies = 5 [(confluent.field_meta).params = {minItems: "2", expression: "size(hobbies) >= 2"}];
repeated Friend friends = 6;
message Address {
option (confluent.message_meta).params = {
expression: "size(address.street) >= 5 && address.street.contains('paris') || address.city == 'paris'"
};
string street = 1 [(confluent.field_meta).params = {minLength: "5", maxLength: "10", expression: "size(street) >= 5 && size(street) <= 10"}];
string city = 2 [(confluent.field_meta).params = {minLength: "2", maxLength: "10"}];
}
message Friend {
string name = 1 [(confluent.field_meta).params = {minLength: "3", maxLength: "10", expression: "size(name) >= 3 && size(name) <= 10"}];
int32 age = 2 [(confluent.field_meta).params = {minimum: "2", maximum: "10", expression: "age >= 2 && age <= 10"}];
}
}
Let's update the schema with our validation rules
- Command
- Output
- Recording
Let's asserts number of registered schemas
Let's produce the same invalid payload again
The payload has been rejected with useful errors
org.apache.kafka.common.errors.PolicyViolationException: Request parameters do not satisfy the configured policy.
Topic 'topic-protobuf' has invalid protobuf schema payload: name length must greater than 2, age must be greater than or equal to 18, Student.name is too short (1 < 3), Student.name does not match expression 'size(name) >= 3 && size(name) <= 50', Student.email does not match format 'email', Student.email does not match expression 'email.contains('foo')', Student.Address.street is too long (56 > 10), Student.Address.street does not match expression 'size(street) >= 5 && size(street) <= 10', Student.Address.city is too short (0 < 2), Student.address does not match expression 'size(address.street) >= 5 && address.street.contains('paris') || address.city == 'paris'', Student.hobbies has too few items (1 < 2), Student.hobbies does not match expression 'size(hobbies) >= 2', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 && age <= 10', Student.Friend.name is too long (11 > 10), Student.Friend.name does not match expression 'size(name) >= 3 && size(name) <= 10', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 && age <= 10'
- Command
- Output
- Recording
cat invalid-payload.json | jq -c | \
kafka-protobuf-console-producer \
--bootstrap-server localhost:6969 \
--producer.config teamA-sa.properties \
--topic topic-protobuf \
--property schema.registry.url=http://localhost:8081 \
--property value.schema.id=2
[2024-04-10 03:58:39,565] INFO KafkaProtobufSerializerConfig values:
auto.register.schemas = true
basic.auth.credentials.source = URL
basic.auth.user.info = [hidden]
bearer.auth.cache.expiry.buffer.seconds = 300
bearer.auth.client.id = null
bearer.auth.client.secret = null
bearer.auth.credentials.source = STATIC_TOKEN
bearer.auth.custom.provider.class = null
bearer.auth.identity.pool.id = null
bearer.auth.issuer.endpoint.url = null
bearer.auth.logical.cluster = null
bearer.auth.scope = null
bearer.auth.scope.claim.name = scope
bearer.auth.sub.claim.name = sub
bearer.auth.token = [hidden]
context.name.strategy = class io.confluent.kafka.serializers.context.NullContextNameStrategy
http.connect.timeout.ms = 60000
http.read.timeout.ms = 60000
id.compatibility.strict = true
key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
latest.cache.size = 1000
latest.cache.ttl.sec = -1
latest.compatibility.strict = true
max.schemas.per.subject = 1000
normalize.schemas = false
proxy.host =
proxy.port = -1
reference.lookup.only = false
reference.subject.name.strategy = class io.confluent.kafka.serializers.subject.DefaultReferenceSubjectNameStrategy
rule.actions = []
rule.executors = []
rule.service.loader.enable = true
schema.format = null
schema.reflection = false
schema.registry.basic.auth.user.info = [hidden]
schema.registry.ssl.cipher.suites = null
schema.registry.ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
schema.registry.ssl.endpoint.identification.algorithm = https
schema.registry.ssl.engine.factory.class = null
schema.registry.ssl.key.password = null
schema.registry.ssl.keymanager.algorithm = SunX509
schema.registry.ssl.keystore.certificate.chain = null
schema.registry.ssl.keystore.key = null
schema.registry.ssl.keystore.location = null
schema.registry.ssl.keystore.password = null
schema.registry.ssl.keystore.type = JKS
schema.registry.ssl.protocol = TLSv1.3
schema.registry.ssl.provider = null
schema.registry.ssl.secure.random.implementation = null
schema.registry.ssl.trustmanager.algorithm = PKIX
schema.registry.ssl.truststore.certificates = null
schema.registry.ssl.truststore.location = null
schema.registry.ssl.truststore.password = null
schema.registry.ssl.truststore.type = JKS
schema.registry.url = [http://localhost:8081]
skip.known.types = true
use.latest.version = false
use.latest.with.metadata = null
use.schema.id = -1
value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
(io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializerConfig:376)
[2024-04-10 03:58:40,601] ERROR Error when sending message to topic topic-protobuf with key: null, value: 119 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback:52)
org.apache.kafka.common.errors.PolicyViolationException: Request parameters do not satisfy the configured policy. Topic 'topic-protobuf' has invalid protobuf schema payload: name length must greater than 2, age must be greater than or equal to 18, Student.name is too short (1 < 3), Student.name does not match expression 'size(name) >= 3 step-17-SH-OUTPUTstep-17-SH-OUTPUT size(name) <= 50', Student.email does not match format 'email', Student.email does not match expression 'email.contains('foo')', Student.Address.street is too long (56 > 10), Student.Address.street does not match expression 'size(street) >= 5 step-17-SH-OUTPUTstep-17-SH-OUTPUT size(street) <= 10', Student.Address.city is too short (0 < 2), Student.address does not match expression 'size(address.street) >= 5 step-17-SH-OUTPUTstep-17-SH-OUTPUT address.street.contains('paris') || address.city == 'paris'', Student.hobbies has too few items (1 < 2), Student.hobbies does not match expression 'size(hobbies) >= 2', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 step-17-SH-OUTPUTstep-17-SH-OUTPUT age <= 10', Student.Friend.name is too long (11 > 10), Student.Friend.name does not match expression 'size(name) >= 3 step-17-SH-OUTPUTstep-17-SH-OUTPUT size(name) <= 10', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 step-17-SH-OUTPUTstep-17-SH-OUTPUT age <= 10'
[2024-04-10 03:58:40,601] ERROR Error when sending message to topic topic-protobuf with key: null, value: 119 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback:52)
org.apache.kafka.common.errors.PolicyViolationException: Request parameters do not satisfy the configured policy. Topic 'topic-protobuf' has invalid protobuf schema payload: name length must greater than 2, age must be greater than or equal to 18, Student.name is too short (1 < 3), Student.name does not match expression 'size(name) >= 3 step-17-SH-OUTPUTstep-17-SH-OUTPUT size(name) <= 50', Student.email does not match format 'email', Student.email does not match expression 'email.contains('foo')', Student.Address.street is too long (56 > 10), Student.Address.street does not match expression 'size(street) >= 5 step-17-SH-OUTPUTstep-17-SH-OUTPUT size(street) <= 10', Student.Address.city is too short (0 < 2), Student.address does not match expression 'size(address.street) >= 5 step-17-SH-OUTPUTstep-17-SH-OUTPUT address.street.contains('paris') || address.city == 'paris'', Student.hobbies has too few items (1 < 2), Student.hobbies does not match expression 'size(hobbies) >= 2', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 step-17-SH-OUTPUTstep-17-SH-OUTPUT age <= 10', Student.Friend.name is too long (11 > 10), Student.Friend.name does not match expression 'size(name) >= 3 step-17-SH-OUTPUTstep-17-SH-OUTPUT size(name) <= 10', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 step-17-SH-OUTPUTstep-17-SH-OUTPUT age <= 10'
Check in the audit log that message was denied
Check in the audit log that message was denied in cluster kafka1
- Command
- Output
- Recording
kafka-console-consumer \
--bootstrap-server localhost:19092,localhost:19093,localhost:19094 \
--topic _conduktor_gateway_auditlogs \
--from-beginning \
--timeout-ms 3000 \| jq 'select(.type=="SAFEGUARD" and .eventData.plugin=="io.conduktor.gateway.interceptor.safeguard.SchemaPayloadValidationPolicyPlugin")'
returns 1 event
{
"id" : "ca499d6a-7fa1-49f1-8a01-a7bf3a939554",
"source" : "krn://cluster=ECCAV3HlS5uwR_NIedQ4Kw",
"type" : "SAFEGUARD",
"authenticationPrincipal" : "teamA",
"userName" : "sa",
"connection" : {
"localAddress" : null,
"remoteAddress" : "/192.168.65.1:30697"
},
"specVersion" : "0.1.0",
"time" : "2024-04-09T23:57:56.097380762Z",
"eventData" : {
"level" : "error",
"plugin" : "io.conduktor.gateway.interceptor.safeguard.SchemaPayloadValidationPolicyPlugin",
"message" : "Request parameters do not satisfy the configured policy. Topic 'topic-protobuf' has invalid protobuf schema payload: name length must greater than 2, age must be greater than or equal to 18, Student.name is too short (1 < 3), Student.name does not match expression 'size(name) >= 3 && size(name) <= 50', Student.email does not match format 'email', Student.email does not match expression 'email.contains('foo')', Student.Address.street is too long (56 > 10), Student.Address.street does not match expression 'size(street) >= 5 && size(street) <= 10', Student.Address.city is too short (0 < 2), Student.address does not match expression 'size(address.street) >= 5 && address.street.contains('paris') || address.city == 'paris'', Student.hobbies has too few items (1 < 2), Student.hobbies does not match expression 'size(hobbies) >= 2', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 && age <= 10', Student.Friend.name is too long (11 > 10), Student.Friend.name does not match expression 'size(name) >= 3 && size(name) <= 10', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 && age <= 10'"
}
}
{"id":"4de07bcd-aa18-4e97-80a1-9d68b273c05e","source":"Optional.empty","type":"REST_API","authenticationPrincipal":"admin","userName":null,"connection":{"localAddress":"172.27.0.7:8888","remoteAddress":"192.168.65.1:42508"},"specVersion":"0.1.0","time":"2024-04-09T23:58:30.367585597Z","eventData":{"method":"POST","path":"/admin/vclusters/v1/vcluster/teamA/username/sa","body":"{\"lifeTimeSeconds\": 7776000}"}}
{"id":"cd234acf-c8e9-4866-87a3-340a350d3724","source":null,"type":"AUTHENTICATION","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":"/172.27.0.7:6969","remoteAddress":"/192.168.65.1:31086"},"specVersion":"0.1.0","time":"2024-04-09T23:58:31.363143875Z","eventData":"SUCCESS"}
{"id":"03e99a92-8a6e-4054-91c8-36d5270ea2df","source":null,"type":"AUTHENTICATION","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":"/172.27.0.7:6970","remoteAddress":"/192.168.65.1:46977"},"specVersion":"0.1.0","time":"2024-04-09T23:58:31.413829500Z","eventData":"SUCCESS"}
{"id":"182848f1-7937-4ee8-baeb-4905c62bae3b","source":null,"type":"AUTHENTICATION","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":"/172.27.0.7:6969","remoteAddress":"/192.168.65.1:31090"},"specVersion":"0.1.0","time":"2024-04-09T23:58:33.494260709Z","eventData":"SUCCESS"}
{"id":"a4707309-6864-4404-8c1e-01c5f0d5bd41","source":null,"type":"AUTHENTICATION","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":"/172.27.0.7:6970","remoteAddress":"/192.168.65.1:46981"},"specVersion":"0.1.0","time":"2024-04-09T23:58:33.538718459Z","eventData":"SUCCESS"}
{"id":"caddb245-6a3a-4002-a76d-c7929fe842c3","source":null,"type":"AUTHENTICATION","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":"/172.27.0.7:6969","remoteAddress":"/192.168.65.1:31092"},"specVersion":"0.1.0","time":"2024-04-09T23:58:34.867617751Z","eventData":"SUCCESS"}
{"id":"9b34d84b-8b79-4ce4-8828-4e9e0758844f","source":null,"type":"AUTHENTICATION","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":"/172.27.0.7:6969","remoteAddress":"/192.168.65.1:31093"},"specVersion":"0.1.0","time":"2024-04-09T23:58:34.896960543Z","eventData":"SUCCESS"}
{"id":"251f9ad1-194c-49f6-952b-67f981384a7d","source":null,"type":"AUTHENTICATION","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":"/172.27.0.7:6970","remoteAddress":"/192.168.65.1:46984"},"specVersion":"0.1.0","time":"2024-04-09T23:58:35.014358501Z","eventData":"SUCCESS"}
{"id":"e39e190b-e21d-402d-b54a-ef30cc90a1ff","source":"Optional.empty","type":"REST_API","authenticationPrincipal":"admin","userName":null,"connection":{"localAddress":"172.27.0.7:8888","remoteAddress":"192.168.65.1:42519"},"specVersion":"0.1.0","time":"2024-04-09T23:58:38.775735670Z","eventData":{"method":"POST","path":"/admin/interceptors/v1/vcluster/teamA/interceptor/guard-schema-payload-validate","body":"{ \"pluginClass\" : \"io.conduktor.gateway.interceptor.safeguard.SchemaPayloadValidationPolicyPlugin\", \"priority\" : 100, \"config\" : { \"schemaRegistryConfig\" : { \"host\" : \"http://schema-registry:8081\" }, \"topic\" : \"topic-.*\", \"schemaIdRequired\" : true, \"validateSchema\" : true, \"action\" : \"BLOCK\" }}"}}
{"id":"d4c36bf3-aa37-43f4-a933-6b33c0b2695e","source":"Optional.empty","type":"REST_API","authenticationPrincipal":"admin","userName":null,"connection":{"localAddress":"172.27.0.7:8888","remoteAddress":"192.168.65.1:42520"},"specVersion":"0.1.0","time":"2024-04-09T23:58:38.993876753Z","eventData":{"method":"GET","path":"/admin/interceptors/v1/vcluster/teamA","body":null}}
{"id":"481e18b8-5a86-4d66-ad82-5deee130df36","source":null,"type":"AUTHENTICATION","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":"/172.27.0.7:6969","remoteAddress":"/192.168.65.1:31125"},"specVersion":"0.1.0","time":"2024-04-09T23:58:40.134200546Z","eventData":"SUCCESS"}
{"id":"d74c7c27-59a4-41f1-a699-587dcda33be7","source":null,"type":"AUTHENTICATION","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":"/172.27.0.7:6970","remoteAddress":"/192.168.65.1:47016"},"specVersion":"0.1.0","time":"2024-04-09T23:58:40.170864629Z","eventData":"SUCCESS"}
{"id":"a551f930-b667-4467-b847-35ab9568e6db","source":"krn://cluster=cF6kjeBGRHqbhzEb4VaZEw","type":"SAFEGUARD","authenticationPrincipal":"teamA","userName":"sa","connection":{"localAddress":null,"remoteAddress":"/192.168.65.1:47016"},"specVersion":"0.1.0","time":"2024-04-09T23:58:40.580200046Z","eventData":{"level":"error","plugin":"io.conduktor.gateway.interceptor.safeguard.SchemaPayloadValidationPolicyPlugin","message":"Request parameters do not satisfy the configured policy. Topic 'topic-protobuf' has invalid protobuf schema payload: name length must greater than 2, age must be greater than or equal to 18, Student.name is too short (1 < 3), Student.name does not match expression 'size(name) >= 3 step-18-AUDITLOG-OUTPUTstep-18-AUDITLOG-OUTPUT size(name) <= 50', Student.email does not match format 'email', Student.email does not match expression 'email.contains('foo')', Student.Address.street is too long (56 > 10), Student.Address.street does not match expression 'size(street) >= 5 step-18-AUDITLOG-OUTPUTstep-18-AUDITLOG-OUTPUT size(street) <= 10', Student.Address.city is too short (0 < 2), Student.address does not match expression 'size(address.street) >= 5 step-18-AUDITLOG-OUTPUTstep-18-AUDITLOG-OUTPUT address.street.contains('paris') || address.city == 'paris'', Student.hobbies has too few items (1 < 2), Student.hobbies does not match expression 'size(hobbies) >= 2', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 step-18-AUDITLOG-OUTPUTstep-18-AUDITLOG-OUTPUT age <= 10', Student.Friend.name is too long (11 > 10), Student.Friend.name does not match expression 'size(name) >= 3 step-18-AUDITLOG-OUTPUTstep-18-AUDITLOG-OUTPUT size(name) <= 10', Student.Friend.age is greater than 10, Student.Friend.age does not match expression 'age >= 2 step-18-AUDITLOG-OUTPUTstep-18-AUDITLOG-OUTPUT age <= 10'"}}
[2024-04-10 03:58:45,120] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.TimeoutException
Processed a total of 13 messages
Let's now produce a valid payload
- Command
- Output
- Recording
cat valid-payload.json | jq -c | \
kafka-protobuf-console-producer \
--bootstrap-server localhost:6969 \
--producer.config teamA-sa.properties \
--topic topic-protobuf \
--property schema.registry.url=http://localhost:8081 \
--property value.schema.id=2
[2024-04-10 03:58:46,139] INFO KafkaProtobufSerializerConfig values:
auto.register.schemas = true
basic.auth.credentials.source = URL
basic.auth.user.info = [hidden]
bearer.auth.cache.expiry.buffer.seconds = 300
bearer.auth.client.id = null
bearer.auth.client.secret = null
bearer.auth.credentials.source = STATIC_TOKEN
bearer.auth.custom.provider.class = null
bearer.auth.identity.pool.id = null
bearer.auth.issuer.endpoint.url = null
bearer.auth.logical.cluster = null
bearer.auth.scope = null
bearer.auth.scope.claim.name = scope
bearer.auth.sub.claim.name = sub
bearer.auth.token = [hidden]
context.name.strategy = class io.confluent.kafka.serializers.context.NullContextNameStrategy
http.connect.timeout.ms = 60000
http.read.timeout.ms = 60000
id.compatibility.strict = true
key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
latest.cache.size = 1000
latest.cache.ttl.sec = -1
latest.compatibility.strict = true
max.schemas.per.subject = 1000
normalize.schemas = false
proxy.host =
proxy.port = -1
reference.lookup.only = false
reference.subject.name.strategy = class io.confluent.kafka.serializers.subject.DefaultReferenceSubjectNameStrategy
rule.actions = []
rule.executors = []
rule.service.loader.enable = true
schema.format = null
schema.reflection = false
schema.registry.basic.auth.user.info = [hidden]
schema.registry.ssl.cipher.suites = null
schema.registry.ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
schema.registry.ssl.endpoint.identification.algorithm = https
schema.registry.ssl.engine.factory.class = null
schema.registry.ssl.key.password = null
schema.registry.ssl.keymanager.algorithm = SunX509
schema.registry.ssl.keystore.certificate.chain = null
schema.registry.ssl.keystore.key = null
schema.registry.ssl.keystore.location = null
schema.registry.ssl.keystore.password = null
schema.registry.ssl.keystore.type = JKS
schema.registry.ssl.protocol = TLSv1.3
schema.registry.ssl.provider = null
schema.registry.ssl.secure.random.implementation = null
schema.registry.ssl.trustmanager.algorithm = PKIX
schema.registry.ssl.truststore.certificates = null
schema.registry.ssl.truststore.location = null
schema.registry.ssl.truststore.password = null
schema.registry.ssl.truststore.type = JKS
schema.registry.url = [http://localhost:8081]
skip.known.types = true
use.latest.version = false
use.latest.with.metadata = null
use.schema.id = -1
value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
(io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializerConfig:376)
And consume it back
- Command
- Output
- Recording
kafka-protobuf-console-consumer \
--bootstrap-server localhost:6969 \
--consumer.config teamA-sa.properties \
--topic topic-protobuf \
--from-beginning \
--timeout-ms 3000
[2024-04-10 03:58:47,643] INFO KafkaProtobufDeserializerConfig values:
auto.register.schemas = true
basic.auth.credentials.source = URL
basic.auth.user.info = [hidden]
bearer.auth.cache.expiry.buffer.seconds = 300
bearer.auth.client.id = null
bearer.auth.client.secret = null
bearer.auth.credentials.source = STATIC_TOKEN
bearer.auth.custom.provider.class = null
bearer.auth.identity.pool.id = null
bearer.auth.issuer.endpoint.url = null
bearer.auth.logical.cluster = null
bearer.auth.scope = null
bearer.auth.scope.claim.name = scope
bearer.auth.sub.claim.name = sub
bearer.auth.token = [hidden]
context.name.strategy = class io.confluent.kafka.serializers.context.NullContextNameStrategy
derive.type = false
http.connect.timeout.ms = 60000
http.read.timeout.ms = 60000
id.compatibility.strict = true
key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
latest.cache.size = 1000
latest.cache.ttl.sec = -1
latest.compatibility.strict = true
max.schemas.per.subject = 1000
normalize.schemas = false
proxy.host =
proxy.port = -1
rule.actions = []
rule.executors = []
rule.service.loader.enable = true
schema.format = null
schema.reflection = false
schema.registry.basic.auth.user.info = [hidden]
schema.registry.ssl.cipher.suites = null
schema.registry.ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
schema.registry.ssl.endpoint.identification.algorithm = https
schema.registry.ssl.engine.factory.class = null
schema.registry.ssl.key.password = null
schema.registry.ssl.keymanager.algorithm = SunX509
schema.registry.ssl.keystore.certificate.chain = null
schema.registry.ssl.keystore.key = null
schema.registry.ssl.keystore.location = null
schema.registry.ssl.keystore.password = null
schema.registry.ssl.keystore.type = JKS
schema.registry.ssl.protocol = TLSv1.3
schema.registry.ssl.provider = null
schema.registry.ssl.secure.random.implementation = null
schema.registry.ssl.trustmanager.algorithm = PKIX
schema.registry.ssl.truststore.certificates = null
schema.registry.ssl.truststore.location = null
schema.registry.ssl.truststore.password = null
schema.registry.ssl.truststore.type = JKS
schema.registry.url = [http://localhost:8081]
specific.protobuf.key.type = class java.lang.Object
specific.protobuf.value.type = class java.lang.Object
use.latest.version = false
use.latest.with.metadata = null
use.schema.id = -1
value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
(io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializerConfig:376)
{"name":"D","age":17,"email":"bad email","address":{"street":"a way too lond adress that will not fit in your database","city":""},"hobbies":["reading"],"friends":[{"name":"Tommy","age":17},{"name":"Emma Watson","age":18}]}
{"name":"Doe","age":18,"email":"foo.doe@example.com","address":{"street":"123 paris","city":"Any paris"},"hobbies":["reading","cycling"],"friends":[{"name":"Tom","age":9},{"name":"Emma","age":10}]}
[2024-04-10 03:58:51,371] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$:44)
org.apache.kafka.common.errors.TimeoutException
[2024-04-10 03:58:51,371] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$:44)
org.apache.kafka.common.errors.TimeoutException
Processed a total of 2 messages
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.
- Command
- Output
- Recording
docker compose down --volumes
Container schema-registry Stopping
Container kafka-client Stopping
Container gateway2 Stopping
Container gateway1 Stopping
Container gateway2 Stopped
Container gateway2 Removing
Container gateway2 Removed
Container gateway1 Stopped
Container gateway1 Removing
Container gateway1 Removed
Container schema-registry Stopped
Container schema-registry Removing
Container schema-registry Removed
Container kafka3 Stopping
Container kafka2 Stopping
Container kafka1 Stopping
Container kafka2 Stopped
Container kafka2 Removing
Container kafka2 Removed
Container kafka1 Stopped
Container kafka1 Removing
Container kafka1 Removed
Container kafka-client Stopped
Container kafka-client Removing
Container kafka-client Removed
Container kafka3 Stopped
Container kafka3 Removing
Container kafka3 Removed
Container zookeeper Stopping
Container zookeeper Stopped
Container zookeeper Removing
Container zookeeper Removed
Network safeguard-validate-schema-payload-proto_default Removing
Network safeguard-validate-schema-payload-proto_default Removed
Conclusion
You can enrich your existing schema to add even more data quality to your systems!