Schema validation in producers in Kafka - Commands & Configuration

Introduction

When sending data to Kafka, producers must ensure the data format is correct. Schema validation helps catch mistakes early by checking data against a defined structure before sending it.

When you want to prevent sending wrong or incomplete data to Kafka topics.

When multiple applications produce data to the same topic and need a consistent format.

When you want to avoid errors in consumers caused by unexpected data shapes.

When evolving data formats over time and need to maintain compatibility.

When using a schema registry to centrally manage data formats.

Config File - producer.properties

producer.properties

bootstrap.servers=localhost:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
schema.registry.url=http://localhost:8081

This configuration file sets up the Kafka producer to connect to the Kafka broker at localhost:9092.

It uses StringSerializer for keys and KafkaAvroSerializer for values, which enables schema validation using Avro format.

The schema.registry.url points to the Schema Registry service that stores and manages schemas.

Commands

This command starts a console producer that sends messages to 'example-topic'. It includes a simple Avro schema for validation with fields 'name' (string) and 'age' (int).

Terminal

kafka-console-producer --broker-list localhost:9092 --topic example-topic --property value.schema='{"type":"record","name":"User","fields":[{"name":"name","type":"string"},{"name":"age","type":"int"}]}'

Expected Output

>

→

--broker-list - Specifies the Kafka broker address to connect to.

→

--topic - Specifies the Kafka topic to send messages to.

→

--property value.schema - Defines the Avro schema for validating message values.

This command sends a valid JSON message matching the schema to the topic. The producer validates the message against the schema before sending.

Terminal

echo '{"name":"Alice","age":30}' | kafka-console-producer --broker-list localhost:9092 --topic example-topic --property value.schema='{"type":"record","name":"User","fields":[{"name":"name","type":"string"},{"name":"age","type":"int"}]}'

Expected Output

No output (command runs silently)

This command tries to send a message where 'age' is a string instead of an int. Schema validation will reject this message.

Terminal

echo '{"name":"Bob","age":"thirty"}' | kafka-console-producer --broker-list localhost:9092 --topic example-topic --property value.schema='{"type":"record","name":"User","fields":[{"name":"name","type":"string"},{"name":"age","type":"int"}]}'

Expected Output

ERROR: Failed to serialize message due to schema validation error

This command reads one message from the beginning of 'example-topic' to verify that only valid messages were sent.

Terminal

kafka-console-consumer --bootstrap-server localhost:9092 --topic example-topic --from-beginning --max-messages 1

Expected Output

{"name":"Alice","age":30}

→

--from-beginning - Reads messages from the start of the topic.

→

--max-messages - Limits the number of messages to read.

Key Concept

If you remember nothing else from this pattern, remember: schema validation in producers stops bad data before it enters Kafka.

Common Mistakes

Not configuring the producer to use the schema registry and serializer.

Without this, the producer cannot validate data and may send invalid messages.

Set 'value.serializer' to KafkaAvroSerializer and provide 'schema.registry.url' in producer config.

Sending messages that do not match the schema fields or types.

Schema validation will reject these messages, causing errors and failed sends.

Ensure message data matches the schema exactly in field names and data types.

Not running a schema registry service or pointing to the wrong URL.

Producer cannot fetch or register schemas, so validation fails.

Run schema registry on localhost:8081 or update 'schema.registry.url' to the correct address.

Summary

Configure the Kafka producer with Avro serializer and schema registry URL for schema validation.

Send messages that match the defined schema to prevent invalid data from entering Kafka.

Use console producer and consumer commands to test schema validation and message flow.

Practice

1. What is the main purpose of schema validation in Kafka producers?

easy

A. To compress messages for storage

B. To speed up message delivery in Kafka

C. To ensure data sent matches the expected format and structure

D. To encrypt messages before sending

Solution

1. Step 1: Understand schema validation role

   Schema validation checks if data matches a predefined format or structure.

2. Step 2: Identify producer's goal

   Producers use schema validation to keep data clean and consistent before sending.

3. Final Answer:

   To ensure data sent matches the expected format and structure -> Option C

4. Quick Check:

   Schema validation = data format check [OK]

Hint: Schema validation means checking data format before sending [OK]

Common Mistakes:

• Confusing validation with encryption
• Thinking it speeds up message delivery
• Assuming it compresses data

2. Which configuration is essential for KafkaAvroSerializer to validate schemas in a producer?

easy

A. bootstrap.servers

B. schema.registry.url

C. acks

D. compression.type

Solution

1. Step 1: Identify serializer requirements

   KafkaAvroSerializer needs to connect to the schema registry to validate schemas.

2. Step 2: Find the required config key

   The key 'schema.registry.url' tells the serializer where to find the schema registry.

3. Final Answer:

   schema.registry.url -> Option B

4. Quick Check:

   Schema registry URL = validation source [OK]

Hint: Set schema.registry.url for schema validation [OK]

Common Mistakes:

• Using bootstrap.servers for schema validation
• Confusing acks with schema config
• Missing schema.registry.url causes validation failure

3. Given this producer code snippet using KafkaAvroSerializer:

props.put("schema.registry.url", "http://localhost:8081");
ProducerRecord record = new ProducerRecord<>("topic", userRecord);
producer.send(record);

What happens if userRecord does not match the registered schema?

medium

A. A serialization exception is thrown and message is not sent

B. The producer retries sending the message indefinitely

C. The message is sent successfully without validation

D. The message is sent but schema registry logs a warning

Solution

1. Step 1: Understand KafkaAvroSerializer behavior

   It validates data against the schema before sending.

2. Step 2: Check what happens on mismatch

   If data doesn't match, serialization fails and an exception is thrown, preventing send.

3. Final Answer:

   A serialization exception is thrown and message is not sent -> Option A

4. Quick Check:

   Schema mismatch = serialization error [OK]

Hint: Mismatch triggers serialization exception, no send [OK]

Common Mistakes:

• Assuming message sends without validation
• Thinking producer retries endlessly
• Believing only warnings are logged

4. You see this error when sending a message: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: 404 Not Found. What is the most likely cause related to schema validation?

medium

A. The schema registry URL is incorrect or unreachable

B. The producer topic does not exist

C. The message key is null

D. The Kafka broker is down

Solution

1. Step 1: Analyze the error message

   404 Not Found means the client can't reach the schema registry endpoint.

2. Step 2: Connect error to schema validation

   If schema registry URL is wrong or server is down, validation fails with this error.

3. Final Answer:

   The schema registry URL is incorrect or unreachable -> Option A

4. Quick Check:

   404 error = schema registry URL issue [OK]

Hint: 404 error means schema registry URL problem [OK]

Common Mistakes:

• Blaming topic existence for 404
• Assuming null key causes this error
• Thinking Kafka broker down causes 404

5. You want to produce messages with optional fields that may or may not be present in your Avro schema. How should you configure your producer and schema to allow schema validation to pass when some fields are missing?

hard

A. Send empty strings for missing fields to pass validation

B. Remove optional fields from the schema and send messages without them

C. Disable schema validation in the producer configuration

D. Define optional fields with default values in the Avro schema and use KafkaAvroSerializer with schema.registry.url set

Solution

1. Step 1: Understand optional fields in Avro

   Optional fields must have default values in the schema to allow missing data.

2. Step 2: Configure producer for validation

   Use KafkaAvroSerializer with schema.registry.url set to validate against schema with defaults.

3. Final Answer:

   Define optional fields with default values in the Avro schema and use KafkaAvroSerializer with schema.registry.url set -> Option D

4. Quick Check:

   Defaults + serializer config = optional fields allowed [OK]

Hint: Use defaults in schema for optional fields [OK]

Common Mistakes:

• Removing fields instead of using defaults
• Disabling validation instead of fixing schema
• Sending empty strings instead of omitting fields