Test Automation

Avro Schema Registry Contract Testing: A Practical Guide

HelpMeTest

HelpMeTest

— 6 min read
Avro Schema Registry Contract Testing: A Practical Guide

Schema evolution is one of the most underappreciated risks in event-driven architectures. A producer team renames a field, removes a required field, or changes a type — and downstream consumers start failing silently or crashing at runtime. Confluent Schema Registry solves the governance problem by enforcing compatibility rules, but only if you test your schemas before deploying.

This guide covers contract testing for Avro schemas with Schema Registry: how to test compatibility modes, how to catch breaking changes before deployment, and how to write integration tests that validate your producers and consumers work with the same schema version.

Why Schema Registry Contract Testing Matters

Without schema contract tests, breaking changes propagate silently:

  • A producer serializes using a new schema with a renamed field
  • The message lands on the Kafka topic
  • The consumer tries to deserialize using the old schema
  • Deserialization either fails (hard crash) or silently uses default values (data corruption)

Schema Registry with compatibility enforcement prevents registration of incompatible schemas. But you need tests to verify:

  1. Your new schema passes the compatibility check before you deploy
  2. Your consumer can deserialize messages produced with both the old and new schema
  3. Your producer correctly registers and uses schemas via the registry

Project Setup

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter'
    implementation 'org.springframework.kafka:spring-kafka'
    implementation 'io.confluent:kafka-avro-serializer:7.6.0'
    implementation 'org.apache.avro:avro:1.11.3'

    testImplementation 'org.testcontainers:kafka:1.19.7'
    testImplementation 'org.testcontainers:junit-jupiter:1.19.7'
    // Schema Registry container
    testImplementation 'com.github.daniel-shuy:kafka-schema-registry-testcontainers:1.2.0'
}

You also need the Confluent Maven repository:

repositories {
    maven { url 'https://packages.confluent.io/maven/' }
}

Defining Avro Schemas

Place Avro schemas in src/main/avro/:

// src/main/avro/OrderEvent.avsc
{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.example.events",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "status", "type": {
      "type": "enum",
      "name": "OrderStatus",
      "symbols": ["CREATED", "PAID", "SHIPPED", "CANCELLED"]
    }},
    {"name": "createdAt", "type": "long", "logicalType": "timestamp-millis"}
  ]
}

A proposed evolution — adding an optional field with a default:

// OrderEvent v2 — backward compatible (new optional field with default)
{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.example.events",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "status", "type": {"type": "enum", "name": "OrderStatus",
      "symbols": ["CREATED", "PAID", "SHIPPED", "CANCELLED"]}},
    {"name": "createdAt", "type": "long", "logicalType": "timestamp-millis"},
    {"name": "region", "type": ["null", "string"], "default": null}
  ]
}

Test Setup: Kafka + Schema Registry

@SpringBootTest
@Testcontainers
public abstract class SchemaRegistryTestBase {

    @Container
    static final KafkaContainer kafka = new KafkaContainer(
            DockerImageName.parse("confluentinc/cp-kafka:7.6.0"))
        .withReuse(true);

    @Container
    static final GenericContainer<?> schemaRegistry = new GenericContainer<>(
            DockerImageName.parse("confluentinc/cp-schema-registry:7.6.0"))
        .withExposedPorts(8081)
        .withEnv("SCHEMA_REGISTRY_HOST_NAME", "schema-registry")
        .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
            "PLAINTEXT://host.docker.internal:9093")
        .dependsOn(kafka)
        .withReuse(true);

    protected static String schemaRegistryUrl;

    @BeforeAll
    static void setupUrls() {
        schemaRegistryUrl = String.format("http://%s:%d",
            schemaRegistry.getHost(),
            schemaRegistry.getMappedPort(8081));
    }

    @DynamicPropertySource
    static void kafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
        registry.add("spring.kafka.properties.schema.registry.url",
            () -> schemaRegistryUrl);
        registry.add("spring.kafka.producer.value-serializer",
            () -> "io.confluent.kafka.serializers.KafkaAvroSerializer");
        registry.add("spring.kafka.consumer.value-deserializer",
            () -> "io.confluent.kafka.serializers.KafkaAvroDeserializer");
        registry.add("spring.kafka.consumer.auto-offset-reset", () -> "earliest");
    }
}

Testing Schema Registration

class SchemaRegistrationTest extends SchemaRegistryTestBase {

    private SchemaRegistryClient registryClient;

    @BeforeEach
    void setUp() {
        registryClient = new CachedSchemaRegistryClient(schemaRegistryUrl, 100);
    }

    @Test
    void shouldRegisterSchemaSuccessfully() throws Exception {
        String schemaJson = new String(Files.readAllBytes(
            Paths.get("src/main/avro/OrderEvent.avsc")));
        Schema schema = new Schema.Parser().parse(schemaJson);

        int schemaId = registryClient.register("orders-value", 
            new AvroSchema(schema));

        assertThat(schemaId).isPositive();
    }

    @Test
    void shouldRetrieveRegisteredSchemaById() throws Exception {
        String schemaJson = new String(Files.readAllBytes(
            Paths.get("src/main/avro/OrderEvent.avsc")));
        Schema schema = new Schema.Parser().parse(schemaJson);

        int id = registryClient.register("orders-value", new AvroSchema(schema));
        ParsedSchema retrieved = registryClient.getSchemaById(id);

        assertThat(retrieved).isNotNull();
        assertThat(retrieved.rawSchema().toString())
            .contains("OrderEvent");
    }
}

Testing Compatibility Modes

This is the core contract test. Before registering a new schema, verify it passes the compatibility check for the current mode:

class SchemaCompatibilityTest extends SchemaRegistryTestBase {

    private SchemaRegistryClient registryClient;
    private static final String SUBJECT = "orders-value";

    @BeforeEach
    void setUp() throws Exception {
        registryClient = new CachedSchemaRegistryClient(schemaRegistryUrl, 100);

        // Set BACKWARD compatibility (new consumers can read old messages)
        registryClient.updateCompatibility(SUBJECT, "BACKWARD");

        // Register v1 schema
        String v1Json = new String(Files.readAllBytes(
            Paths.get("src/main/avro/OrderEvent.avsc")));
        registryClient.register(SUBJECT, new AvroSchema(new Schema.Parser().parse(v1Json)));
    }

    @Test
    void shouldAllowBackwardCompatibleSchemaEvolution() throws Exception {
        // v2 adds an optional field with default — backward compatible
        String v2Json = loadSchema("src/test/avro/OrderEventV2.avsc");
        Schema v2Schema = new Schema.Parser().parse(v2Json);

        boolean compatible = registryClient.testCompatibility(
            SUBJECT, new AvroSchema(v2Schema));

        assertThat(compatible).isTrue();
    }

    @Test
    void shouldRejectRemovalOfRequiredField() throws Exception {
        // Schema that removes "customerId" — backward incompatible
        String breakingSchemaJson = """
            {
              "type": "record",
              "name": "OrderEvent",
              "namespace": "com.example.events",
              "fields": [
                {"name": "orderId", "type": "string"},
                {"name": "amount", "type": "double"},
                {"name": "createdAt", "type": "long"}
              ]
            }
            """;
        Schema breakingSchema = new Schema.Parser().parse(breakingSchemaJson);

        boolean compatible = registryClient.testCompatibility(
            SUBJECT, new AvroSchema(breakingSchema));

        assertThat(compatible).isFalse();
    }

    @Test
    void shouldRejectFieldTypeChange() throws Exception {
        // Changing "amount" from double to string — incompatible
        String typeChangeJson = """
            {
              "type": "record",
              "name": "OrderEvent",
              "namespace": "com.example.events",
              "fields": [
                {"name": "orderId", "type": "string"},
                {"name": "customerId", "type": "string"},
                {"name": "amount", "type": "string"},
                {"name": "createdAt", "type": "long"}
              ]
            }
            """;

        boolean compatible = registryClient.testCompatibility(
            SUBJECT, new AvroSchema(new Schema.Parser().parse(typeChangeJson)));

        assertThat(compatible).isFalse();
    }

    @Test
    void shouldRejectAddingRequiredFieldWithoutDefault() throws Exception {
        // Adding a required field (no default) breaks backward compatibility
        String requiredFieldJson = """
            {
              "type": "record",
              "name": "OrderEvent",
              "namespace": "com.example.events",
              "fields": [
                {"name": "orderId", "type": "string"},
                {"name": "customerId", "type": "string"},
                {"name": "amount", "type": "double"},
                {"name": "createdAt", "type": "long"},
                {"name": "requiredNewField", "type": "string"}
              ]
            }
            """;

        boolean compatible = registryClient.testCompatibility(
            SUBJECT, new AvroSchema(new Schema.Parser().parse(requiredFieldJson)));

        assertThat(compatible).isFalse();
    }
}

Testing Producer-Consumer with Schema Evolution

This is the end-to-end contract test — does a consumer using the new schema correctly read messages produced with the old schema?

class SchemaEvolutionEndToEndTest extends SchemaRegistryTestBase {

    @Autowired
    private KafkaTemplate<String, GenericRecord> kafkaTemplate;

    @Test
    void shouldConsumeV1MessageWithV2Consumer() throws Exception {
        // Producer registers and uses v1 schema
        Schema v1Schema = loadAvroSchema("src/main/avro/OrderEvent.avsc");
        GenericRecord v1Message = new GenericData.Record(v1Schema);
        v1Message.put("orderId", "ORD-001");
        v1Message.put("customerId", "CUST-123");
        v1Message.put("amount", 149.99d);
        v1Message.put("status", new GenericData.EnumSymbol(
            v1Schema.getField("status").schema(), "CREATED"));
        v1Message.put("createdAt", System.currentTimeMillis());

        kafkaTemplate.send("orders", "ORD-001", v1Message).get();

        // Consumer using v2 schema (with new optional "region" field)
        Schema v2Schema = loadAvroSchema("src/test/avro/OrderEventV2.avsc");
        Map<String, Object> consumerProps = buildConsumerProps();
        consumerProps.put("schema.registry.url", schemaRegistryUrl);

        KafkaConsumer<String, GenericRecord> consumer = 
            new KafkaConsumer<>(consumerProps);
        consumer.subscribe(List.of("orders"));

        ConsumerRecords<String, GenericRecord> records = 
            consumer.poll(Duration.ofSeconds(10));

        assertThat(records.count()).isEqualTo(1);

        GenericRecord received = records.iterator().next().value();
        // Core fields present
        assertThat(received.get("orderId").toString()).isEqualTo("ORD-001");
        assertThat(received.get("customerId").toString()).isEqualTo("CUST-123");
        // New optional field has default value (null)
        assertThat(received.get("region")).isNull();

        consumer.close();
    }
}

Testing FULL Compatibility Mode

For stricter environments, FULL compatibility requires both forward and backward compatibility:

@Test
void shouldEnforceFullCompatibilityMode() throws Exception {
    registryClient.updateCompatibility("strict-orders-value", "FULL");

    // v1 schema
    registryClient.register("strict-orders-value", new AvroSchema(v1Schema));

    // A change that is backward-compatible but NOT forward-compatible:
    // Adding an optional field IS backward compatible but NOT forward compatible
    // because old consumers cannot read the new field.
    // Under FULL mode, this should fail.
    Schema backwardOnlySchema = new Schema.Parser().parse("""
        {
          "type": "record",
          "name": "StrictOrderEvent",
          "namespace": "com.example.events",
          "fields": [
            {"name": "orderId", "type": "string"},
            {"name": "newOptionalField", "type": ["null", "string"], "default": null}
          ]
        }
        """);

    // Under FULL mode: adding optional field is backward compatible
    // but an old consumer reading a new message would not know about the field
    // This test documents your compatibility mode behavior
    boolean compatible = registryClient.testCompatibility(
        "strict-orders-value", new AvroSchema(backwardOnlySchema));

    // FULL compatibility allows optional fields with defaults
    // because old consumers will simply ignore unknown fields
    assertThat(compatible).isTrue();
}

CI Integration: Schema Compatibility Gate

The most impactful use of these tests is as a CI gate — run before deploying a service to prevent breaking changes from reaching production:

/**
 * This test acts as a pre-deployment gate.
 * It will fail if the current schema in src/main/avro/ is incompatible
 * with the schema already registered in the staging Schema Registry.
 */
@Test
void currentSchemaMustBeCompatibleWithRegisteredSchema() throws Exception {
    String stagingRegistryUrl = System.getenv("STAGING_SCHEMA_REGISTRY_URL");
    Assumptions.assumeTrue(stagingRegistryUrl != null,
        "Skipping — STAGING_SCHEMA_REGISTRY_URL not set");

    SchemaRegistryClient stagingClient = 
        new CachedSchemaRegistryClient(stagingRegistryUrl, 100);

    String currentSchemaJson = new String(Files.readAllBytes(
        Paths.get("src/main/avro/OrderEvent.avsc")));
    Schema currentSchema = new Schema.Parser().parse(currentSchemaJson);

    boolean compatible = stagingClient.testCompatibility(
        "orders-value", new AvroSchema(currentSchema));

    assertThat(compatible)
        .as("Current OrderEvent schema is not compatible with staging registry. " +
            "Cannot deploy until schema compatibility is resolved.")
        .isTrue();
}

Testing Serialization Round-Trip

A basic sanity test that catches serialization bugs — serialize an object, deserialize it, verify the values:

@Test
void shouldSerializeAndDeserializeAvroMessageRoundTrip() throws Exception {
    Schema schema = loadAvroSchema("src/main/avro/OrderEvent.avsc");

    // Serialize
    GenericRecord original = new GenericData.Record(schema);
    original.put("orderId", "ORD-ROUNDTRIP");
    original.put("customerId", "CUST-999");
    original.put("amount", 299.99d);
    original.put("status", new GenericData.EnumSymbol(
        schema.getField("status").schema(), "PAID"));
    original.put("createdAt", 1717200000000L);

    ByteArrayOutputStream out = new ByteArrayOutputStream();
    DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(schema);
    Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
    writer.write(original, encoder);
    encoder.flush();

    // Deserialize
    DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
    Decoder decoder = DecoderFactory.get().binaryDecoder(out.toByteArray(), null);
    GenericRecord deserialized = reader.read(null, decoder);

    // Assert
    assertThat(deserialized.get("orderId").toString()).isEqualTo("ORD-ROUNDTRIP");
    assertThat(deserialized.get("amount")).isEqualTo(299.99d);
    assertThat(deserialized.get("createdAt")).isEqualTo(1717200000000L);
}

Key Takeaways

Schema contract testing with Avro and Schema Registry has four essential components:

  • Compatibility tests — call testCompatibility() before every schema registration. This is the CI gate that prevents breaking changes from being deployed.
  • Round-trip serialization tests — verify serialize/deserialize produces identical values, especially for complex types like enums and unions.
  • End-to-end evolution tests — confirm that a consumer using the new schema can correctly read messages produced with the old schema.
  • Compatibility mode documentation tests — codify which compatibility mode (BACKWARD, FORWARD, FULL) applies to each subject, and write tests that verify the boundary cases.

The CI schema gate test is the highest-leverage addition. It makes schema incompatibility a build failure, not a production incident.

Read more

Start now free