Schema validation in producers in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka producer schema validation code?

Consider a Kafka producer configured with Avro schema validation. The producer tries to send a message that does not conform to the registered schema. What will happen?

Kafka

from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer

schema_str = '''{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "int"}
  ]
}'''

value_schema = avro.loads(schema_str)

producer_config = {
    'bootstrap.servers': 'localhost:9092',
    'schema.registry.url': 'http://localhost:8081'
}

producer = AvroProducer(producer_config, default_value_schema=value_schema)

try:
    producer.produce(topic='users', value={'name': 'Alice', 'age': 'twenty'})
    producer.flush()
    print('Message sent successfully')
except Exception as e:
    print(f'Error: {e}')

A Error: KeyError: 'age'

B Message sent successfully

C RuntimeError: Producer not connected

D Error: <class 'confluent_kafka.avro.serializer.MessageSerializerError'>: Value for field age is not an int

💡 Hint

Check the data types in the message against the schema fields.

The schema expects the field 'age' to be an integer, but the message provides a string 'twenty'. This causes a serialization error during schema validation.

🧠 Conceptual

intermediate

Which option correctly describes schema validation in Kafka producers?

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

A To encrypt messages before sending to Kafka topics

B To ensure messages conform to a predefined schema before sending to Kafka topics

C To compress messages to reduce network usage

D To authenticate the producer with Kafka brokers

💡 Hint

Think about why schemas are used in data pipelines.

Schema validation ensures that the data sent by producers matches the expected format, preventing corrupt or incompatible data from entering Kafka topics.

🔧 Debug

advanced

What error does this Kafka producer code raise?

Identify the error raised by this Kafka producer code snippet when schema validation is enabled.

Kafka

from confluent_kafka.avro import AvroProducer
from confluent_kafka import avro

producer_config = {
    'bootstrap.servers': 'localhost:9092',
    'schema.registry.url': 'http://localhost:8081'
}

value_schema_str = '''{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "int"}
  ]
}'''

value_schema = avro.loads(value_schema_str)

producer = AvroProducer(producer_config, default_value_schema=value_schema)

producer.produce(topic='users', value={'name': 'Bob'})
producer.flush()

A MessageSerializerError: Missing required field 'age'

B TypeError: value must be a dict

C No error, message sent successfully

D KeyError: 'age'

💡 Hint

Check if all required fields in the schema are present in the message.

The schema requires both 'name' and 'age' fields. The message is missing 'age', so schema validation fails with a MessageSerializerError.

📝 Syntax

advanced

Which option correctly configures schema validation for a Kafka Avro producer?

Choose the correct Python code snippet to configure an AvroProducer with schema validation.

A producer = AvroProducer({'bootstrap.servers': 'localhost:9092'}, default_value_schema=value_schema)

B producer = AvroProducer({'bootstrap.servers': 'localhost:9092', 'schema.registry.url': 'http://localhost:8081'}, value_schema=value_schema)

C producer = AvroProducer({'bootstrap.servers': 'localhost:9092', 'schema.registry.url': 'http://localhost:8081'}, default_value_schema=value_schema)

D producer = AvroProducer({'schema.registry.url': 'http://localhost:8081'}, default_value_schema=value_schema)

💡 Hint

Check the required keys in the configuration dictionary and parameter names.

The AvroProducer requires both 'bootstrap.servers' and 'schema.registry.url' in the config. The schema must be passed as default_value_schema.

🚀 Application

expert

How many messages will be successfully sent with this Kafka producer code?

Given the following code, how many messages will be sent successfully to the 'users' topic?

Kafka

from confluent_kafka.avro import AvroProducer
from confluent_kafka import avro

schema_str = '''{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "int"}
  ]
}'''

value_schema = avro.loads(schema_str)

producer_config = {
    'bootstrap.servers': 'localhost:9092',
    'schema.registry.url': 'http://localhost:8081'
}

producer = AvroProducer(producer_config, default_value_schema=value_schema)

messages = [
    {'name': 'Alice', 'age': 30},
    {'name': 'Bob', 'age': 'thirty'},
    {'name': 'Charlie', 'age': 25},
    {'name': 'Dana'}
]

success_count = 0

for msg in messages:
    try:
        producer.produce(topic='users', value=msg)
        success_count += 1
    except Exception:
        pass
producer.flush()
print(success_count)

A 2

B 4

C 3

D 1

💡 Hint

Check which messages conform to the schema and which cause errors.

Only messages with both 'name' as string and 'age' as int pass validation. 'Bob' has age as string, 'Dana' is missing 'age'. So only 'Alice' and 'Charlie' succeed.

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