10.4.4 – Troubleshooting Schema Issues

Module: Section 10 – Kafka Ecosystem & Tools
Section: 10.4 – Schema Registry Operations
Goal: Diagnose and resolve common Schema Registry issues in production

Introduction

This module provides systematic troubleshooting workflows for common Schema Registry problems.

Why This Module Matters: Fast incident resolution prevents extended outages and data loss.

Issue 1: Producer SerializationException

Symptoms

org.apache.kafka.common.errors.SerializationException:
  Error serializing Avro message
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException:
  Schema being registered is incompatible with an earlier schema; error code: 409

Root Causes

1. Incompatible schema change (most common)
2. Wrong compatibility mode
3. Subject naming strategy mismatch

Troubleshooting Steps

Step 1: Check current schema

curl http://localhost:8081/subjects/users-value/versions/latest | jq '.schema'

Step 2: Test compatibility

curl -X POST http://localhost:8081/compatibility/subjects/users-value/versions/latest \
  -d '{"schema": "..."}' | jq '.'

Response:

{
  "is_compatible": false,
  "messages": ["Reader field email not found in writer schema"]
}

Step 3: Fix schema

// ❌ Bad: Required field added
{"name": "email", "type": "string"}

// ✅ Good: Optional field with default
{"name": "email", "type": ["null", "string"], "default": null}

Quick Fix (Temporary)

Disable compatibility temporarily (⚠️ use with caution):

# Set to NONE temporarily
curl -X PUT http://localhost:8081/config/users-value \
  -d '{"compatibility": "NONE"}'

# Register schema
curl -X POST http://localhost:8081/subjects/users-value/versions \
  -d '{"schema": "..."}'

# Re-enable compatibility
curl -X PUT http://localhost:8081/config/users-value \
  -d '{"compatibility": "BACKWARD"}'

Issue 2: Consumer DeserializationException

Symptoms

org.apache.kafka.common.errors.SerializationException:
  Error deserializing Avro message
Caused by: java.io.EOFException: Reached end of input

Root Causes

1. Corrupted message
2. Wrong schema ID
3. Schema Registry unreachable
4. Deleted schema

Troubleshooting Steps

Step 1: Check Schema Registry connectivity

curl http://localhost:8081/subjects

Step 2: Inspect message hex

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic users \
  --max-messages 1 \
  --formatter kafka.tools.DefaultMessageFormatter \
  --property print.value=true \
  --property print.key=false

Expected format: [0x00][schema_id (4 bytes)][avro_data]

Step 3: Extract schema ID

# First byte should be 0x00 (magic byte)
# Next 4 bytes are schema ID (big-endian)

Step 4: Verify schema exists

curl http://localhost:8081/schemas/ids/42

If 404: Schema was deleted → permanent data loss!

Quick Fix

Skip corrupted message:

Properties props = new Properties();
props.put("auto.offset.reset", "earliest");

while (true) {
    try {
        ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
        // Process records
    } catch (SerializationException e) {
        logger.error("Skipping corrupted message: {}", e.getMessage());
        consumer.seek(consumer.assignment().iterator().next(), consumer.position() + 1);
    }
}

Issue 3: Schema Registry Unreachable

Symptoms

io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException:
  Connection refused; error code: 50001

Root Causes

1. Schema Registry down
2. Network issue
3. Wrong URL configured
4. Firewall blocking port 8081

Troubleshooting Steps

Step 1: Check if Schema Registry is running

curl http://localhost:8081/

Step 2: Check Schema Registry logs

tail -f /var/log/schema-registry/schema-registry.log

Step 3: Verify network connectivity

telnet localhost 8081

Step 4: Check producer/consumer configuration

props.put("schema.registry.url", "http://localhost:8081");  // ✅ Correct
// vs
props.put("schema.registry.url", "http://localhost:9092");  // ❌ Wrong (Kafka broker port)

Quick Fix

Use cached schemas temporarily:

// Producer: Cache schema ID to avoid registry calls
private static final Map<Schema, Integer> schemaCache = new ConcurrentHashMap<>();

int schemaId = schemaCache.computeIfAbsent(schema, s -> {
    return schemaRegistry.register(subject, s);
});

Issue 4: Primary Instance Not Elected

Symptoms

HTTP 422: This node is not the primary and can't accept write requests

Root Causes

1. All instances in standby mode
2. ZooKeeper connection lost
3. _schemas topic unavailable

Troubleshooting Steps

Step 1: Check instance status

curl http://schema-registry-1:8081/mode
curl http://schema-registry-2:8081/mode
curl http://schema-registry-3:8081/mode

Expected: One instance returns {"mode": "READWRITE"}, others return standby indicators.

Step 2: Check ZooKeeper connectivity

echo "ruok" | nc localhost 2181

Step 3: Check _schemas topic

kafka-topics --describe --topic _schemas --bootstrap-server localhost:9092

Step 4: Check Schema Registry logs

INFO  Elected as primary

Quick Fix

Force primary election (⚠️ dangerous):

# Restart all Schema Registry instances
# Primary will be re-elected

Issue 5: Deleted Schema Breaks Consumers

Symptoms

Schema not found for ID 42

Root Causes

1. Schema permanently deleted
2. _schemas topic data loss
3. Schema Registry cache cleared

Troubleshooting Steps

Step 1: Check if schema exists

curl http://localhost:8081/schemas/ids/42

Response:

{
  "error_code": 40403,
  "message": "Schema not found"
}

Step 2: Check _schemas topic

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic _schemas \
  --from-beginning \
  --property print.key=true

Step 3: Restore from backup

# If you have schema backup
curl -X POST http://localhost:8081/subjects/users-value/versions \
  -d '{"schema": "..."}'

Prevention

Never permanently delete schemas in production:

# ✅ Soft delete (recoverable)
curl -X DELETE http://localhost:8081/subjects/users-value

# ❌ Permanent delete (IRREVERSIBLE!)
curl -X DELETE "http://localhost:8081/subjects/users-value?permanent=true"

Backup strategy:

# Daily schema backup
0 2 * * * /usr/local/bin/backup-schemas.sh

Issue 6: Slow Schema Registry Performance

Symptoms

• P99 latency > 100ms
• Producer/consumer slowdowns
• Timeouts

Root Causes

1. Low cache hit rate
2. Small heap size
3. _schemas topic replication lag
4. Disk I/O bottleneck

Troubleshooting Steps

Step 1: Check cache hit rate

# JMX metric
kafka.schema.registry:type=cache-hit-rate

Expected: > 95%

Step 2: Check heap usage

jstat -gc <schema-registry-pid>

Step 3: Check _schemas topic lag

kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group schema-registry \
  --describe

Quick Fix

Increase cache size:

# schema-registry.properties
max.schemas.per.subject=10000  # Default: 1000

Increase heap:

export SCHEMA_REGISTRY_HEAP_OPTS="-Xms2g -Xmx2g"

Common Error Codes

Troubleshooting Checklist

When Producer Fails

• Check Schema Registry URL in producer config
• Test schema compatibility locally
• Verify Schema Registry is reachable
• Check compatibility mode for subject
• Review producer logs for full error

When Consumer Fails

• Verify Schema Registry connectivity
• Check if schema ID exists
• Inspect message format (hex dump)
• Check consumer logs for serialization errors
• Verify _schemas topic is healthy

When Schema Registry Unavailable

• Check Schema Registry process is running
• Verify network connectivity (telnet/curl)
• Check ZooKeeper connectivity
• Verify _schemas topic exists
• Check if primary instance is elected

Detailed Error Scenarios

Error 1: Schema Not Found

Full Error Message:

org.apache.kafka.common.errors.SerializationException: 
  Error deserializing key/value for partition users-0 at offset 1234
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: 
  Schema not found for id 42

What It Means: The schema ID embedded in the message does not exist in Schema Registry.

Typical Causes:

1. Schema Registry is down or unreachable
2. Schema ID was deleted from Registry
3. Message produced by different cluster/Schema Registry
4. Network partition between consumer and Schema Registry
5. Consumer configured with wrong Schema Registry URL

5-Step Debugging Workflow:

# Step 1: Verify Schema Registry is accessible
curl http://localhost:8081/subjects
# Expected: List of subjects (200 OK)

# Step 2: Check if schema ID exists
curl http://localhost:8081/schemas/ids/42
# Expected: Schema definition or 40403 error

# Step 3: Inspect message header manually
kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic users \
  --property print.key=true \
  --max-messages 1 | xxd
# Look for: [0x00][0x00 0x00 0x00 0x2A] at start

# Step 4: List all schema IDs in registry
curl http://localhost:8081/schemas/ids | jq '.'

# Step 5: Check consumer configuration
grep "schema.registry.url" consumer.properties

Solutions:

• ✅ Fix Schema Registry connectivity
• ✅ Re-register missing schema (if you have definition)
• ✅ Verify consumer points to correct Schema Registry
• ✅ Check network ACLs and firewall rules

Error 2: Incompatible Schema Error

Full Error Message:

io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: 
  Schema compatibility check failed
  error code: 409

What It Means: New schema violates compatibility rules configured in Schema Registry.

Typical Causes:

1. Removed required field without default
2. Changed field type incompatibly (e.g., int → string)
3. Breaking backward/forward compatibility rules
4. Wrong compatibility mode for use case

Debugging Example:

# Check current compatibility mode
curl http://localhost:8081/config/users-value | jq '.'
# Response: {"compatibilityLevel": "BACKWARD"}

# Test new schema compatibility
curl -X POST http://localhost:8081/compatibility/subjects/users-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"email\",\"type\":\"string\"}]}"
  }' | jq '.'

# Response shows why it's incompatible:
{
  "is_compatible": false,
  "messages": [
    "Reader field 'email' not found in writer schema",
    "Field 'email' is required but has no default value"
  ]
}

Solutions:

• ✅ Add default value to new fields: {"name": "email", "type": ["null", "string"], "default": null}
• ✅ Don't remove required fields without default
• ✅ Use new field names instead of changing types
• ✅ Review compatibility mode (BACKWARD vs FORWARD vs FULL)

Error 3: ClassCastException / Serialization Errors

Full Error Message:

org.apache.kafka.common.errors.SerializationException: 
  Error deserializing key/value
Caused by: java.lang.ClassCastException: 
  org.apache.avro.generic.GenericData$Record cannot be cast to com.example.User

OR

org.apache.avro.AvroTypeException: Found int, expecting string

What It Means: Mismatch between expected Java class and schema/data structure.

Typical Causes:

1. Using wrong generated classes or outdated schemas
2. Manually creating records that don't conform to schema
3. Schema evolution changed field types without proper handling
4. Consumer expects SpecificRecord but receives GenericRecord
5. Field type changed (e.g., int to string) between producer/consumer versions

5-Step Debugging Workflow:

# Step 1: Check schema used by producer
curl http://localhost:8081/subjects/users-value/versions/latest | jq '.schema | fromjson'

# Step 2: Check consumer's generated code version
grep "version" target/generated-sources/avro/com/example/User.java

# Step 3: Verify consumer deserializer config
grep "specific.avro.reader" consumer.properties
# Should be: specific.avro.reader=true

# Step 4: Inspect actual message data type
kafka-avro-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic users \
  --property schema.registry.url=http://localhost:8081 \
  --max-messages 1

# Step 5: Check for recent schema changes
curl http://localhost:8081/subjects/users-value/versions | jq '.'

Solutions:

• ✅ Regenerate Java classes: mvn clean compile
• ✅ Ensure specific.avro.reader=true in consumer config
• ✅ Validate producer and consumer use compatible schema versions
• ✅ Check for type mismatches in schema evolution
• ✅ Use GenericRecord if schema changes frequently

Complete Debugging Example

Scenario: Consumer fails with AvroTypeException: Found int, expecting string

Full Debugging Workflow:

Step 1: Check the schema registered for the topic's subject

curl http://localhost:8081/subjects/users-value/versions/latest | jq '.schema | fromjson'

Step 2: Verify Avro schema used by producer matches or is compatible

# Producer code review
cat src/main/avro/user.avsc | jq '.fields[] | select(.name=="age")'

Step 3: Ensure consumer's generated classes expect correct schema

# Check generated Java class
grep -A 5 "private.*age" target/generated-sources/avro/com/example/User.java

Step 4: Confirm producer not sending older/incompatible schema

# Check producer logs for schema registration
grep "Schema registered" producer.log

Step 5: Look for recent schema changes

# Get all versions
curl http://localhost:8081/subjects/users-value/versions | jq '.[]'

# Compare V1 vs V2
diff <(curl -s http://localhost:8081/subjects/users-value/versions/1 | jq '.schema | fromjson') \
     <(curl -s http://localhost:8081/subjects/users-value/versions/2 | jq '.schema | fromjson')

Resolution: Found that V2 changed age field from int to string without updating consumers.

Fix: Add new field ageStr with default, keep old age field for backward compatibility.

Troubleshooting Checklist

Defensive Coding Best Practices

1. Always Handle Serialization Exceptions Gracefully

❌ Bad:

producer.send(record);  // Crashes if serialization fails

✅ Good:

try {
    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            log.error("Failed to send record", exception);
            // Send to DLQ or retry queue
            dlqProducer.send(new ProducerRecord<>("dlq-topic", record));
        }
    });
} catch (SerializationException e) {
    log.error("Schema validation failed", e);
    metrics.incrementCounter("schema_errors");
}

2. Use Schema Validation Tools During Build

<!-- Maven plugin to validate schemas at build time -->
<plugin>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-schema-registry-maven-plugin</artifactId>
    <executions>
        <execution>
            <goals>
                <goal>test-compatibility</goal>
            </goals>
        </execution>
    </executions>
</plugin>

3. Implement Retry Policies for Transient Errors

RetryPolicy<Object> retryPolicy = new RetryPolicy<>()
    .handle(RestClientException.class)
    .withDelay(Duration.ofSeconds(1))
    .withMaxRetries(3);

Failsafe.with(retryPolicy).run(() -> {
    schemaRegistryClient.register(subject, schema);
});

4. Use Schema References and Documented Contracts

{
  "type": "record",
  "name": "Order",
  "doc": "Order event - see https://wiki.company.com/order-schema",
  "fields": [
    {
      "name": "orderId",
      "type": "string",
      "doc": "Unique order identifier - UUID format required"
    }
  ]
}

5. Consider Schema-Aware Deserialization

When schema evolves frequently:

// Use GenericRecord for flexibility
props.put("specific.avro.reader", "false");

ConsumerRecord<String, GenericRecord> record = ...;
GenericRecord value = record.value();

// Safely access fields
Object email = value.hasField("email") ? value.get("email") : null;

6. Monitor Topics with Schema Visualization Tools

Tools:

• Kafka UI (formerly kafka-ui-lite)
• Confluent Control Center
• Kafdrop
• AKHQ (kafka HQ)

What to Monitor:

• Schema versions per topic
• Compatibility mode changes
• Failed schema registrations
• Schema deletion events

7. Implement Circuit Breakers

CircuitBreaker breaker = CircuitBreaker.ofDefaults("schemaRegistry");

// Wrap Schema Registry calls
Try<Integer> result = CircuitBreaker.decorateTry(breaker, () -> 
    schemaRegistryClient.register(subject, schema)
).get();

if (result.isFailure()) {
    // Fallback: Use cached schema or default behavior
    log.warn("Schema Registry unavailable, using cached schema");
}

Key Takeaways

1. SerializationException: Usually incompatible schema change. Test compatibility before deploying.

2. DeserializationException: Check schema ID exists, inspect message format, verify Schema Registry connectivity.

3. Connection Refused: Schema Registry down or wrong URL configured.

4. 422 Standby: Request sent to standby instance. Route to primary.

5. Deleted Schema: Permanent deletion breaks consumers. Use soft delete only.

6. Slow Performance: Check cache hit rate, heap size, _schemas topic lag.

7. Backup Strategy: Daily schema backups prevent data loss.

8. Never Delete Permanently: Soft delete allows recovery.

9. Monitoring: Proactive monitoring prevents most issues.

10. Compatibility Testing: Always test locally before production.

What's Next?

Next Module: 10.5 – ksqlDB Fundamentals

What We Covered:

• ✅ Common Schema Registry issues
• ✅ Systematic troubleshooting workflows
• ✅ Quick fixes and prevention
• ✅ Error code reference

What's Coming:

• ❌ Introduction to ksqlDB
• ❌ Streams vs Tables in ksqlDB
• ❌ ksqlDB architecture

References:

• REST API → 10.4.2: Schema Registry REST API
• Monitoring → 10.4.3: Monitoring Schema Registry
• Schema evolution → 10.3: Schema Evolution & Compatibility