Olay Şemasının Evrimi — Alanın Adını Değiştirdin, Kim Kırıldı?

Bir bankanın havale formu var ve şubelerin bir kısmı hâlâ eski formu okuyor. Formun altına isteğe bağlı yeni bir kutu eklemek kimseyi şaşırtmaz. Ama “tutar” kutusunun adını değiştirirsen, eski formu okuyan şube tutarı bulamaz.

Olay kimin?

Bir olay Kafka’ya yazıldığında onu birçok servis okuyabilir: bildirim, muhasebe, raporlama. Her biri kendi zamanında yayına çıkar; hepsi aynı gün güncellenmez.

Bu yüzden olayın biçimi bir sözleşmedir. Üreticinin kendi testleri yeşil olsa da, eski kodla okuyan bir tüketici kırılabilir.

Hangi değişiklik kırar?

Varsayılanı olan yeni bir alan eklemek geriye uyumludur?Terimin tanımını açgeriye uyumlulukYeni şemanın eski şemayla yazılmış verileri okuyabilmesi. Varsayılanı olan alan eklemek uyumludur; ad ya da tür değiştirmek değildir.Sözlükte gör →: eski kod onu görmezden gelir. Bir alanın adını ya da türünü değiştirmek ise uyumlu değildir.

Bir schema registry?Terimin tanımını açschema registryOlay şemalarının sürümlerini saklayan ve yeni bir şemayı eskileriyle karşılaştıran servis. Uyumsuz şemayı kaydetmez, böylece yayınlanamaz.Sözlükte gör →, yeni şemayı eskileriyle karşılaştırır. Uyumsuz şemayı kaydetmez; üretici olayı yayınlamadan hata alır.

Kendin gör

1. Varsayılanla oynat. Ad değişti, registry kapalı: bildirim servisi 0 TL yazdı, muhasebe defteri durdu.
2. “Tür değişti” seç. İki tüketici de okuyamadı.
3. Registry kontrolünü aç. Kırıcı şema reddedildi; olay hiç yayınlanmadı, kimse kırılmadı.
4. “Varsayılanı olan yeni alan” seç. Registry açık da olsa kapalı da olsa kimse kırılmadı.

Tuzaklar

Adı tek seferde değiştirmek. Önce yeni alanı ekle ve bir süre ikisini birlikte yaz. Bütün tüketiciler yenisine geçince eskisini kaldır.

Bilinmeyen alanda hata veren tüketici. Tüketici, tanımadığı bir alan görünce hata verirse, en güvenli değişiklik olan alan eklemek bile onu kırar. Okurken hoşgörülü ol.

Eski olayları unutmak. Topic’te aylar önceki olaylar durur. Yeniden işlemek gerektiğinde yeni kod onları da okuyabilmelidir.

1
{
2
  "type": "record",
3
  "name": "TransferCompleted",
4
  "namespace": "com.bank.transfer.events",
5
  "fields": [
6
    { "name": "transferId", "type": "string" },
7
    { "name": "amount", "type": "string", "doc": "Decimal as text, e.g. 500.00" },
8
    { "name": "currency", "type": "string" },
9
    { "name": "channel", "type": "string", "default": "MOBILE", "doc": "Added in v2 with a default, so v1 readers are unaffected" }
10
  ]
11
}

1
spring:
2
  kafka:
3
    producer:
4
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
5
    properties:
6
      schema.registry.url: ${SCHEMA_REGISTRY_URL}
7
      # Never register a schema from the app: register it in CI, where an
8
      # incompatible change fails the build instead of a publish.
9
      auto.register.schemas: false
10
      use.latest.version: true

1
@Component
2
class TransferCompletedListener {
3

4
    private final SmsSender sms;
5

6
    TransferCompletedListener(SmsSender sms) {
7
        this.sms = sms;
8
    }
9

10
    @KafkaListener(topics = "transfer-completed", groupId = "notification")
11
    void on(TransferCompleted event) {
12
        // Fail loudly instead of telling a customer "0 TL".
13
        BigDecimal amount = new BigDecimal(Objects.requireNonNull(event.getAmount(), "amount missing"));
14
        sms.send(event.getTransferId(), "Hesabınıza " + amount + " " + event.getCurrency() + " geldi.");
15
    }
16
}

# Runs in CI before deploy: is the new schema compatible with the latest one?
curl --fail -s -X POST \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data "{\"schema\": $(jq -Rs . < src/main/avro/TransferCompleted.avsc)}" \
  "$SCHEMA_REGISTRY_URL/compatibility/subjects/transfer-completed-value/versions/latest" \
  | jq -e '.is_compatible == true'

Kendini sına