Kafka Consumer Fails To Process Messages Due To Dimension Mismatch

Q: Debug – Investigation Steps 1. Verify the schema registered for the topic

curl -s http://schema-registry:8081/subjects/nlp-embeddings-value/versions/latest | jq . { "subject": "nlp-embeddings-value", "version": 3, "id": 45, "schema": "{...\"fields\":[{\"name\":\"embedding\",\"type\":{\"type\":\"fixed\",\"size\":512}}...]}" } 2. Inspect a sample message payload kafka-avro-console-consumer \ --bootstrap-server broker:9092 \ --topic nlp-embeddings \ --from-beginning \ --max-messages 1 \ --property schema.registry.url=http://schema-registry:8081 The consumer threw the same SerializationException, confirming the mismatch.

Q: 3. Compare producer’s Avro schema

# producer_schema.avsc { "type": "record", "name": "EmbeddingRecord", "fields": [ {"name":"id","type":"string"}, {"name":"embedding","type":{"type":"fixed","size":768}} // new size ] } 4. Review consumer configuration properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092"); properties.put(ConsumerConfig.GROUP_ID_CONFIG, "nlp-consumer"); properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroDeserializer"); properties.put("schema.registry.url", "http://schema-registry:8081"); 5. Check compatibility mode in Schema Registry curl -s http://schema-registry:8081/config | jq . { "compatibilityLevel": "BACKWARD" } Solution – Align Schemas and Guard Deserialization 1. Register a compatible schema version Because the new model emits a larger vector, the safest approach is to evolve the schema to a bytes type with a logical annotation, allowing variable length while preserving backward compatibility.

Q: 2. Update producer to use the new schema

Schema.Parser parser = new Schema.Parser(); Schema schema = parser.parse(new File("new_schema.avsc")); GenericRecord record = new GenericData.Record(schema); record.put("id", uuid); float[] vec = model.encode(text); // 768‑dimensional ByteBuffer buf = ByteBuffer.allocate(vec.length * 4); for (float f : vec) buf.putFloat(f); record.put("embedding", buf.array()); producer.send(new ProducerRecord("nlp-embeddings", key, record)); 3. Add defensive checks in the custom deserializer (optional) If the team prefers to keep the fixed‑size definition, introduce a wrapper that validates length before delegating to Avro deserializer.

In this article

Table of Contents

Problem – Consumer Crashes on Embedding Dimension Mismatch

In a staging deployment of an AI inference pipeline (Kafka 3.3 brokers, Confluent Schema Registry 7.x), the downstream consumer that reads nlp-embeddings messages began throwing SerializationException and shutting down its poll loop. The failure manifested as:

org.apache.kafka.common.errors.SerializationException: Size of data 3072 does not match expected fixed size 2048 for field 'embedding'.
InvalidRecordException: field 'embedding' has incompatible type – expected fixed[512] but found fixed[768].

Consequences:

Consumer lag spiked to >10 minutes.
Upstream producer continued to publish, filling the topic.
Downstream model serving stalled, causing SLA breaches.

Root Cause – Schema Evolution Breaks Fixed‑Size Vector Definition

The producer switched to a new model version that emits 768‑dimensional float vectors. The Avro schema registered in Schema Registry still defined the embedding field as fixed[512] (or bytes with a fixed length). According to the Kafka documentation on Schema Evolution and Compatibility, a change from fixed[512] to fixed[768] is incompatible under the default BACKWARD compatibility setting. The deserializer, configured via value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer, validates the incoming byte array against the registered schema and aborts when the length does not match.

Additional contributing factors observed in the real incident logs:

RecordTooLargeException appeared after the model upgrade because the larger payload exceeded max.request.size.
CI pipeline reported “Incompatible schema: field ’embedding’ changed from fixed[512] to fixed[1024]”.

Debug – Investigation Steps

1. Verify the schema registered for the topic

curl -s http://schema-registry:8081/subjects/nlp-embeddings-value/versions/latest | jq .
{
  "subject": "nlp-embeddings-value",
  "version": 3,
  "id": 45,
  "schema": "{...\"fields\":[{\"name\":\"embedding\",\"type\":{\"type\":\"fixed\",\"size\":512}}...]}"
}

2. Inspect a sample message payload

kafka-avro-console-consumer \
  --bootstrap-server broker:9092 \
  --topic nlp-embeddings \
  --from-beginning \
  --max-messages 1 \
  --property schema.registry.url=http://schema-registry:8081

The consumer threw the same SerializationException, confirming the mismatch.

3. Compare producer’s Avro schema

# producer_schema.avsc
{
  "type": "record",
  "name": "EmbeddingRecord",
  "fields": [
    {"name":"id","type":"string"},
    {"name":"embedding","type":{"type":"fixed","size":768}}  // new size
  ]
}

4. Review consumer configuration

properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "nlp-consumer");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
               "org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
               "io.confluent.kafka.serializers.KafkaAvroDeserializer");
properties.put("schema.registry.url", "http://schema-registry:8081");

5. Check compatibility mode in Schema Registry

curl -s http://schema-registry:8081/config | jq .
{
  "compatibilityLevel": "BACKWARD"
}

Solution – Align Schemas and Guard Deserialization

1. Register a compatible schema version

Because the new model emits a larger vector, the safest approach is to evolve the schema to a bytes type with a logical annotation, allowing variable length while preserving backward compatibility.

# new_schema.avsc
{
  "type": "record",
  "name": "EmbeddingRecord",
  "fields": [
    {"name":"id","type":"string"},
    {
      "name":"embedding",
      "type":{
        "type":"bytes",
        "logicalType":"float_vector",
        "metadata":{"description":"float32 array, variable length"}
      }
    }
  ]
}

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "$(cat new_schema.avsc)"}' \
  http://schema-registry:8081/subjects/nlp-embeddings-value/versions

2. Update producer to use the new schema

Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(new File("new_schema.avsc"));
GenericRecord record = new GenericData.Record(schema);
record.put("id", uuid);
float[] vec = model.encode(text); // 768‑dimensional
ByteBuffer buf = ByteBuffer.allocate(vec.length * 4);
for (float f : vec) buf.putFloat(f);
record.put("embedding", buf.array());

producer.send(new ProducerRecord<>("nlp-embeddings", key, record));

3. Add defensive checks in the custom deserializer (optional)

If the team prefers to keep the fixed‑size definition, introduce a wrapper that validates length before delegating to Avro deserializer.

public class SafeEmbeddingDeserializer implements Deserializer<EmbeddingRecord> {
    private final KafkaAvroDeserializer inner = new KafkaAvroDeserializer();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        inner.configure(configs, isKey);
    }

    @Override
    public EmbeddingRecord deserialize(String topic, byte[] data) {
        // Avro header is 5 bytes; extract payload size
        int expected = 512 * 4; // 512 floats * 4 bytes
        if (data.length - 5 != expected) {
            throw new SerializationException(
                "Size of data " + (data.length - 5) +
                " does not match expected fixed size " + expected + " for field 'embedding'.");
        }
        return (EmbeddingRecord) inner.deserialize(topic, data);
    }

    @Override public void close() { inner.close(); }
}

Update consumer config:

properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
               "com.example.SafeEmbeddingDeserializer");

Verify – Confirm the Fix

Consume a fresh message after redeploying producer and consumer.
Observe that no SerializationException appears in the consumer logs.
Check consumer lag returns to near‑zero:

kafka-consumer-groups --bootstrap-server broker:9092 \
  --describe --group nlp-consumer
GROUP           TOPIC            PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
nlp-consumer    nlp-embeddings   0          1250            1250            0

Run a downstream integration test that feeds a known sentence and asserts the embedding vector length matches the model’s output.
Inspect Schema Registry version history to ensure the new schema is marked BACKWARD_TRANSITIVE.

Prevent – Operational Guardrails

Guardrail	Implementation
Enforce schema compatibility checks in CI	Run `curl -X POST …/subjects/.../versions` with `--fail` and fail the build on non‑compatible responses.
Validate embedding size before publishing	Add a pre‑publish hook that asserts `embedding.length == expectedSize` or logs a warning if the model version changed.
Monitor payload size metrics	Alert when `kafka.producer.record.size.max` exceeds 80 % of `max.request.size` for the `nlp-embeddings` topic.
Versioned schema per model release	Use subject naming strategy `TopicNameStrategy` with model version suffix (e.g., `nlp-embeddings-v2-value`) to isolate breaking changes.
Automated deserializer sanity checks	Deploy the `SafeEmbeddingDeserializer` in production to catch future dimension drifts early.

Related Topic Hub: Data Infrastructure Troubleshooting Hub

FAQ

Why does the consumer crash only after a model upgrade? The new model emits a longer vector, violating the fixed‑size Avro definition. Existing consumers still expect the old size, causing a SerializationException.
Can I keep using a fixed‑size field and avoid schema changes? Only if you guarantee that all future models emit the same dimension. Otherwise, switch to a variable‑length bytes type with logical annotations.
What compatibility mode should I use for future embedding schema changes? BACKWARD_TRANSITIVE or FULL_TRANSITIVE with a bytes field allows adding new dimensions without breaking existing consumers.
How do I detect a payload that exceeds max.request.size before it reaches the broker? Enable the producer metric request-size-max and set an alert on spikes; also add a client‑side size check before producer.send().
Is there a way to automatically migrate old messages to the new schema? Use a Kafka Streams job or ksqlDB to read the old topic, deserialize with the old schema, re‑serialize with the new schema (padding or truncating vectors as needed), and write to a new topic.