05 - Schema Registry: Avro, Protobuf, Compatibilita e Schema Evolution

Il giorno in cui qualcuno rinomina user_id in userId

Venti team scrivono su cinquanta topic. Un giorno un team cambia il campo user_id in userId perché «è più coerente con il resto del codice», e i consumatori che leggono quel topic non si rompono con un errore: continuano a girare e producono dati sbagliati, perché il campo che cercano è sempre vuoto. Kafka non ha colpa: trasporta byte, non sa che cosa significhino. Serve un contratto sulla forma dei messaggi, controllato da una macchina nel momento in cui qualcuno cerca di cambiarlo.

Lo Schema Registry di Confluent fa tre cose. Custodisce gli schemi, ciascuno con un identificatore numerico. Fa scrivere quell'identificatore nei primi byte di ogni messaggio. E, quando registri una nuova versione di uno schema, la confronta con le precedenti secondo una regola di compatibilità che scegli tu. In questo articolo vedi le tre cose con codice eseguito: l'Avro vero (libreria avro-1.12.2.jar, SHA-1 verificato) per la serializzazione e per il controllo di compatibilità, e Protobuf (protobuf-java 3.25.5) per i numeri di campo. Il Registry in sé non c'è: nei listati lo sostituisce una mappa in memoria, e lo dichiaro dove serve. Le regole e le impostazioni vengono dalla documentazione Confluent «Schema Evolution and Compatibility» e «Formats, Serializers, and Deserializers», consultate il 2026-11-12.

Come arriva lo schema giusto al consumatore: il formato del messaggio

Il serializzatore non scrive lo schema in ogni messaggio (sarebbe uno spreco) ma un riferimento. La documentazione descrive il wire format: il byte 0 è la versione del formato (vale 0), i byte da 1 a 4 sono l'identificatore dello schema come intero a 4 byte in ordine big-endian, e dal byte 5 in poi c'è il dato serializzato (per Protobuf si aggiunge, dopo l'ID, un array di indici del tipo di messaggio). Il consumatore legge l'ID, chiede lo schema al Registry (e lo tiene in cache), e decodifica con quello. Il Registry usa un topic Kafka, per default _schemas, con una sola partizione, come log su cui scrive schemi, metadati e impostazioni di compatibilità.

import org.apache.avro.Schema;
import org.apache.avro.generic.*;
import org.apache.avro.io.*;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.*;

public class FormatoFilo {
    static final Map<Integer, Schema> registro = new HashMap<>();   // modello del Registry: id -> schema

    static byte[] serializza(int idSchema, Schema schema, GenericRecord r) throws IOException {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        out.write(0);                                               // byte 0: versione del formato
        out.write(ByteBuffer.allocate(4).putInt(idSchema).array()); // byte 1-4: id dello schema, big-endian
        BinaryEncoder enc = EncoderFactory.get().binaryEncoder(out, null);
        new GenericDatumWriter<GenericRecord>(schema).write(r, enc);
        enc.flush();
        return out.toByteArray();
    }

    static GenericRecord deserializza(byte[] msg, Schema lettore) throws IOException {
        ByteBuffer b = ByteBuffer.wrap(msg);
        b.get();                                                    // salta la versione
        Schema scrittore = registro.get(b.getInt());                // lo schema con cui è stato scritto
        BinaryDecoder dec = DecoderFactory.get().binaryDecoder(msg, 5, msg.length - 5, null);
        return new GenericDatumReader<GenericRecord>(scrittore, lettore).read(null, dec);
    }

    static String hex(byte[] m, int n) {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < n; i++) sb.append(String.format("%02x ", m[i]));
        return sb.toString().trim();
    }

    public static void main(String[] args) throws IOException {
        String base = "{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"}";
        Schema v1 = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Order\",\"fields\":[" + base + "]}");
        Schema v2 = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Order\",\"fields\":[" + base
            + ",{\"name\":\"coupon\",\"type\":[\"null\",\"string\"],\"default\":null}]}");
        registro.put(1, v1);
        registro.put(2, v2);

        GenericRecord vecchio = new GenericData.Record(v1);
        vecchio.put("order_id", "A-1");
        vecchio.put("amount", 19.9);
        byte[] m1 = serializza(1, v1, vecchio);
        System.out.println("messaggio v1: " + m1.length + " byte, prefisso " + hex(m1, 5));
        System.out.println("letto con lo schema v2: " + deserializza(m1, v2));

        GenericRecord nuovo = new GenericData.Record(v2);
        nuovo.put("order_id", "A-2");
        nuovo.put("amount", 5.0);
        nuovo.put("coupon", "SALDI");
        byte[] m2 = serializza(2, v2, nuovo);
        System.out.println("messaggio v2: " + m2.length + " byte, prefisso " + hex(m2, 5));
        System.out.println("letto con lo schema v1: " + deserializza(m2, v1));
    }
}

// Output:
// messaggio v1: 17 byte, prefisso 00 00 00 00 01
// letto con lo schema v2: {"order_id": "A-1", "amount": 19.9, "coupon": null}
// messaggio v2: 24 byte, prefisso 00 00 00 00 02
// letto con lo schema v1: {"order_id": "A-2", "amount": 5.0}

Il metodo serializza scrive esattamente il prefisso della documentazione: uno zero, poi l'ID in 4 byte. Il primo messaggio pesa 17 byte con prefisso 00 00 00 00 01; il secondo, che ha il campo coupon, ne pesa 24 con prefisso 00 00 00 00 02. La parte interessante è la lettura: deserializza recupera dal registro lo schema con cui il messaggio è stato scritto e lo affianca allo schema con cui il consumatore vuole leggere. Con lo schema v2 il vecchio messaggio produce coupon: null, il valore predefinito; con lo schema v1 il nuovo messaggio perde coupon, che il lettore non conosce. È la risoluzione degli schemi di Avro, ed è ciò che il consumatore reale fa per te. Il Registry vero aggiunge la cache, la rete, l'autenticazione e il controllo di compatibilità, che il modello omette.

Le regole di compatibilità, verificate con Avro

La documentazione Confluent definisce i livelli così: BACKWARD (il default) significa che i consumatori che usano lo schema nuovo possono leggere i dati scritti con il vecchio; FORWARD che i consumatori con lo schema vecchio leggono i dati scritti con il nuovo; FULL entrambe le cose. Le varianti TRANSITIVE confrontano con tutte le versioni precedenti e non solo con l'ultima. Il motivo per cui BACKWARD è il default, secondo la stessa documentazione, è che permette di riavvolgere i consumatori fino all'inizio del topic e rileggere i dati vecchi con lo schema nuovo.

Le definizioni si fissano meglio provandole. Il listato usa il controllo di compatibilità della libreria Avro, con lo stesso criterio di direzione: BACKWARD vuol dire «lettore = schema nuovo, scrittore = schema vecchio», FORWARD il contrario. Sei modifiche tipiche dello schema Order:

import org.apache.avro.Schema;
import org.apache.avro.SchemaCompatibility;
import org.apache.avro.SchemaCompatibility.SchemaCompatibilityType;

public class CompatibilitaAvro {
    static Schema schema(String campi) {
        return new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Order\",\"fields\":[" + campi + "]}");
    }

    // BACKWARD: il nuovo schema legge dati scritti con il vecchio. FORWARD: il vecchio legge dati scritti col nuovo.
    static boolean legge(Schema lettore, Schema scrittore) {
        return SchemaCompatibility.checkReaderWriterCompatibility(lettore, scrittore).getType()
            == SchemaCompatibilityType.COMPATIBLE;
    }

    static void prova(String cambio, Schema vecchio, Schema nuovo) {
        boolean bw = legge(nuovo, vecchio), fw = legge(vecchio, nuovo);
        System.out.printf("%-42s BACKWARD=%-5s FORWARD=%-5s FULL=%s%n", cambio, bw, fw, bw && fw);
    }

    public static void main(String[] args) {
        String id = "{\"name\":\"order_id\",\"type\":\"string\"}";
        String importo = "{\"name\":\"amount\",\"type\":\"double\"}";
        String valuta = "{\"name\":\"currency\",\"type\":\"string\",\"default\":\"EUR\"}";
        Schema v1 = schema(id + "," + importo + "," + valuta);

        prova("aggiungo campo con default", v1,
            schema(id + "," + importo + "," + valuta + ",{\"name\":\"coupon\",\"type\":[\"null\",\"string\"],\"default\":null}"));
        prova("aggiungo campo SENZA default", v1,
            schema(id + "," + importo + "," + valuta + ",{\"name\":\"note\",\"type\":\"string\"}"));
        prova("tolgo un campo che aveva il default", v1, schema(id + "," + importo));
        prova("tolgo un campo senza default", v1, schema(id + "," + valuta));
        prova("amount da double a string", v1,
            schema(id + ",{\"name\":\"amount\",\"type\":\"string\"}," + valuta));
        prova("amount da int a long (promozione)",
            schema(id + ",{\"name\":\"amount\",\"type\":\"int\"}," + valuta),
            schema(id + ",{\"name\":\"amount\",\"type\":\"long\"}," + valuta));
    }
}

// Output:
// aggiungo campo con default                 BACKWARD=true  FORWARD=true  FULL=true
// aggiungo campo SENZA default               BACKWARD=false FORWARD=true  FULL=false
// tolgo un campo che aveva il default        BACKWARD=true  FORWARD=true  FULL=true
// tolgo un campo senza default               BACKWARD=true  FORWARD=false FULL=false
// amount da double a string                  BACKWARD=false FORWARD=false FULL=false
// amount da int a long (promozione)          BACKWARD=true  FORWARD=false FULL=false

Leggile riga per riga. Aggiungere un campo con default è compatibile in tutte le direzioni; senza default rompe BACKWARD, perché il lettore nuovo non saprebbe che valore dare al campo mancante nei dati vecchi. Togliere un campo che aveva un default è compatibile in tutte le direzioni; togliere un campo senza default passa BACKWARD (il lettore nuovo ignora il dato in più) ma rompe FORWARD (il lettore vecchio cerca un campo che non c'è e non ha un valore di riserva). Cambiare amount da double a string rompe ovunque. La promozione da int a long è l'unico cambio di tipo che passa, e solo in BACKWARD, perché un lettore a 64 bit capisce un intero a 32 bit ma non viceversa. La tabella della documentazione Confluent per Avro è coerente con questi risultati; la libreria Avro non è però il Registry, e il confronto con il comportamento del Registry vero su un cluster non l'ho eseguito.

Da queste regole discende l'ordine di aggiornamento. Con BACKWARD aggiorni prima i consumatori e poi i produttori: il consumatore nuovo legge sia i dati vecchi sia quelli nuovi. Con FORWARD si aggiornano prima i produttori, e la documentazione aggiunge una condizione: i dati prodotti con gli schemi vecchi non devono più essere leggibili dai consumatori. Con FULL non devi decidere, ma sei limitato ai campi opzionali, cioè con default. Per Kafka Streams la documentazione ammette solo i livelli FULL, TRANSITIVE e BACKWARD. Il livello NONE spegne il controllo e va tenuto fuori dalla produzione.

Protobuf: la compatibilità sta nei numeri

In Protobuf un campo è identificato sul filo dal suo numero, non dal nome. Aggiungere un campo con un numero nuovo è sicuro; rinominarlo non cambia nulla per chi legge i byte; riusare un numero per un'altra cosa è il guaio. Il modello costruisce tre descrittori dello stesso messaggio Order con la libreria Protobuf ufficiale e passa i byte tra versioni diverse.

import com.google.protobuf.*;
import com.google.protobuf.DescriptorProtos.*;
import com.google.protobuf.DescriptorProtos.FieldDescriptorProto.Type;
import com.google.protobuf.Descriptors.Descriptor;

public class NumeriProtobuf {
    // Costruisce al volo un messaggio Order con i campi indicati (nome, numero, tipo).
    static Descriptor ordine(Object[][] campi) throws Exception {
        DescriptorProto.Builder msg = DescriptorProto.newBuilder().setName("Order");
        for (Object[] c : campi)
            msg.addField(FieldDescriptorProto.newBuilder().setName((String) c[0]).setNumber((Integer) c[1])
                .setType((Type) c[2]).setLabel(FieldDescriptorProto.Label.LABEL_OPTIONAL));
        FileDescriptorProto file = FileDescriptorProto.newBuilder().setName("order.proto").setSyntax("proto3")
            .addMessageType(msg).build();
        return Descriptors.FileDescriptor.buildFrom(file, new Descriptors.FileDescriptor[0]).findMessageTypeByName("Order");
    }

    public static void main(String[] args) throws Exception {
        Descriptor v1 = ordine(new Object[][] {{"order_id", 1, Type.TYPE_STRING}, {"amount", 3, Type.TYPE_DOUBLE}});
        Descriptor v2 = ordine(new Object[][] {{"order_id", 1, Type.TYPE_STRING}, {"amount", 3, Type.TYPE_DOUBLE},
                                               {"currency", 4, Type.TYPE_STRING}});
        Descriptor v3 = ordine(new Object[][] {{"order_id", 1, Type.TYPE_STRING}, {"amount", 3, Type.TYPE_STRING}});

        byte[] scrittoV1 = DynamicMessage.newBuilder(v1)
            .setField(v1.findFieldByName("order_id"), "A-1").setField(v1.findFieldByName("amount"), 19.9).build().toByteArray();
        byte[] scrittoV2 = DynamicMessage.newBuilder(v2)
            .setField(v2.findFieldByName("order_id"), "A-2").setField(v2.findFieldByName("amount"), 5.0)
            .setField(v2.findFieldByName("currency"), "EUR").build().toByteArray();

        DynamicMessage a = DynamicMessage.parseFrom(v2, scrittoV1);
        System.out.println("dati v1 letti con v2: currency=\"" + a.getField(v2.findFieldByName("currency")) + "\" (valore predefinito)");

        DynamicMessage b = DynamicMessage.parseFrom(v1, scrittoV2);
        System.out.println("dati v2 letti con v1: campi sconosciuti conservati = " + b.getUnknownFields().asMap().keySet());

        DynamicMessage c = DynamicMessage.parseFrom(v3, scrittoV1);
        System.out.println("numero 3 riusato con un altro tipo: amount=\"" + c.getField(v3.findFieldByName("amount"))
            + "\", campi sconosciuti = " + c.getUnknownFields().asMap().keySet());
    }
}

// Output:
// dati v1 letti con v2: currency="" (valore predefinito)
// dati v2 letti con v1: campi sconosciuti conservati = [4]
// numero 3 riusato con un altro tipo: amount="", campi sconosciuti = [3]

Nella prima riga i dati scritti con v1 letti da v2 danno currency="", il valore predefinito di una stringa. Nella seconda i dati v2 letti da v1 non fanno danni: il campo 4, che v1 non conosce, viene conservato tra i «campi sconosciuti». Nella terza il numero 3, che era un double, è stato riusato per una string: il dato non viene rifiutato, amount risulta vuoto e il valore finisce tra i campi sconosciuti. Nessuna eccezione, nessun avviso, mentre la guida al linguaggio dice chiaramente che «field numbers should never be reused». È l'errore silenzioso che il Registry può bloccare solo se la compatibilità è attiva. Per Protobuf la documentazione Confluent consiglia di usare BACKWARD_TRANSITIVE, perché aggiungere nuovi tipi di messaggio non è compatibile in avanti.

Avro, Protobuf o JSON Schema: come scegliere

La tabella non dice «il migliore»: dice dove sta il contratto in ciascun formato e dove sbagli senza accorgertene. La scelta più solida segue un criterio semplice: usa il formato che i tuoi team già conoscono e che la tua catena di strumenti già legge, e metti in pratica la compatibilità prima del formato. Non do cifre sul risparmio di byte tra i tre formati: dipende dai tuoi dati, e l'unico modo di saperlo è serializzare un campione reale con tutti e tre.

Governance: nomi, registrazione e controllo in CI

La strategia predefinita dei nomi dei subject è TopicNameStrategy: il subject è <topic>-value per i valori e <topic>-key per le chiavi (value.subject.name.strategy, key.subject.name.strategy). Con questa scelta uno schema vive con il suo topic, il che va bene finché in un topic c'è un solo tipo di evento. Per mescolare più tipi nello stesso topic la documentazione indica di spegnere la registrazione automatica e di usare i riferimenti tra schemi.

Sulla registrazione automatica la documentazione è esplicita: in produzione conviene spegnere auto.register.schemas e accendere use.latest.version, per evitare «Schema not found» e registrazioni accidentali; lo schema lo registra la pipeline di rilascio, non il primo produttore che parte. Il controllo vero sta allora in CI: prima del merge si chiede al Registry se il nuovo schema è compatibile con l'ultima versione del subject. Lo schema dello script che segue è lo strumento: la risposta dell'endpoint contiene il campo booleano is_compatible e il livello applicato è quello configurato per il subject (altrimenti quello globale).

# Schema da adattare (non eseguito: serve un Registry raggiungibile).
# Controlla se uno schema è compatibile con l'ultima versione del subject, con il livello configurato per quel subject.
SCHEMA_JSON=$(jq -Rs '{schema: .}' src/main/avro/Order.avsc)

curl -s -X POST "http://schema-registry-staging:8081/compatibility/subjects/orders-value/versions/latest" \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d "$SCHEMA_JSON"
# risposta attesa dal manuale: un oggetto con il campo booleano "is_compatible"

Il comando non è stato eseguito perché non c'è un Registry in questo ambiente; la forma della richiesta e il campo is_compatible vengono dalla documentazione dell'API REST. Se la chiamata restituisce false, la pipeline deve fallire: è esattamente il momento in cui un user_id rinominato viene fermato, prima di arrivare sul topic. Ricorda anche che il Registry tiene i dati nel topic _schemas: replica e retention di quel topic fanno parte della tua strategia di backup.

Che cosa sai fare adesso

Sai che cosa c'è nei primi cinque byte di un messaggio con schema e perché il consumatore può leggere dati scritti con una versione diversa dalla sua. Sai prevedere se una modifica di uno schema Avro passa in BACKWARD, FORWARD o FULL, e in quale ordine aggiornare produttori e consumatori. Sai che in Protobuf il rischio sta nei numeri di campo e che riusarli non produce errori. Sai che in produzione lo schema si registra dalla pipeline e si controlla in CI. Per la parte del flusso che porta questi messaggi verso i database continua con Kafka Connect; per le impostazioni del producer che li invia, torna ai producer avanzati.