Event Schema Evolution

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

// Conceptual schema registry architecture
 
interface SchemaRegistry {
  // Subject: Logical grouping of schema versions (e.g., "order-created")
  // Schema: Definition in specific format (Avro, Protobuf, JSON Schema)
  // SchemaId: Globally unique identifier
  // Version: Sequential version within subject
  
  // Core CRUD operations
  registerSchema(subject: string, schema: Schema): Promise<RegisterResult>;
  getSchema(schemaId: number): Promise<Schema>;
  getSchemaByVersion(subject: string, version: number): Promise<Schema>;
  getLatestSchema(subject: string): Promise<Schema>;
  listVersions(subject: string): Promise<VersionInfo[]>;
  listSubjects(): Promise<string[]>;
  
  // Compatibility operations
  checkCompatibility(subject: string, newSchema: Schema): Promise<CompatResult>;
  getCompatibilityConfig(subject: string): Promise<CompatibilityMode>;
  setCompatibilityConfig(subject: string, mode: CompatibilityMode): Promise<void>;
  
  // Lifecycle operations  
  deleteSchema(subject: string, version: number): Promise<void>;
  deleteSubject(subject: string): Promise<void>;
}
 
interface RegisterResult {
  schemaId: number;      // Globally unique ID
  version: number;       // Version within subject
  isNew: boolean;        // false if identical schema already exists
}
 
interface VersionInfo {
  version: number;
  schemaId: number;
  createdAt: Date;
  isDeprecated: boolean;
}
 
enum CompatibilityMode {
  NONE = 'NONE',                        // No compatibility check
  BACKWARD = 'BACKWARD',                // New can read old
  BACKWARD_TRANSITIVE = 'BACKWARD_TRANSITIVE',
  FORWARD = 'FORWARD',                  // Old can read new
  FORWARD_TRANSITIVE = 'FORWARD_TRANSITIVE',
  FULL = 'FULL',                        // Both directions
  FULL_TRANSITIVE = 'FULL_TRANSITIVE',  // Both directions, all versions
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

// Design-time: Register schema during CI/CD pipeline
// gradle/maven task or npm script
 
// build.gradle.kts
plugins {
    id("com.github.imflog.kafka-schema-registry-gradle-plugin")
}
 
schemaRegistry {
    url.set("http://schema-registry:8081")
    
    register {
        subject("order-created-value", "src/main/avro/OrderCreated.avsc")
        subject("order-updated-value", "src/main/avro/OrderUpdated.avsc")
    }
    
    compatibility {
        subject("order-created-value", "src/main/avro/OrderCreated.avsc")
    }
}
 
// CI/CD pipeline
// 1. ./gradlew schemaRegistryCompatibilityCheck  (fail if incompatible)
// 2. ./gradlew schemaRegistryRegister            (register new version)
// 3. Deploy application
 
// Application uses known schema ID
const SCHEMA_ID = process.env.ORDER_CREATED_SCHEMA_ID;  // 42
 
async function publishOrder(order: Order) {
  const encoded = avroEncode(order, await getSchemaById(SCHEMA_ID));
  await kafka.send({
    topic: 'orders.created',
    messages: [{
      key: order.id,
      value: encoded,
      headers: { 'schema-id': SCHEMA_ID.toString() },
    }],
  });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

// Runtime: Register schema on first message
// Simpler setup but less control
 
import { SchemaRegistry, AvroSerializer } from '@kafkajs/confluent-schema-registry';
 
const registry = new SchemaRegistry({
  host: 'http://schema-registry:8081',
});
 
// Serializer auto-registers schema if not exists
const serializer = new AvroSerializer(registry, {
  autoRegisterSchemas: true,  // Registers on first use
  subject: 'order-created-value',
});
 
async function publishOrder(order: Order) {
  // Serializer embeds schema ID in payload magic byte
  const encoded = await serializer.serialize(order, schema);
  
  await kafka.send({
    topic: 'orders.created',
    messages: [{
      key: order.id,
      value: encoded,  // Magic byte + schema ID + Avro data
    }],
  });
}
 
// Warning: Runtime registration risks
// - No compatibility check before production
// - Schema must be identical across producer instances
// - Network failures during registration block publishing

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// Consumer: Deserialize with schema from registry
import { SchemaRegistry, AvroDeserializer } from '@kafkajs/confluent-schema-registry';
 
const registry = new SchemaRegistry({
  host: 'http://schema-registry:8081',
  // Cache schemas to avoid repeated fetches
  schemaCache: new Map(),
  idCache: new Map(),
});
 
const deserializer = new AvroDeserializer(registry);
 
await consumer.run({
  eachMessage: async ({ message }) => {
    // Deserializer reads schema ID from message prefix
    // Fetches schema from registry (cached after first fetch)
    // Deserializes Avro data using schema
    const order = await deserializer.deserialize(message.value);
    
    await processOrder(order);
  },
});
 
// Under the hood:
// 1. Read magic byte (0x00) and schema ID (4 bytes) from message prefix
// 2. Check cache for schema ID
// 3. If not cached, fetch GET /schemas/ids/{id}
// 4. Cache schema for future messages
// 5. Deserialize remaining bytes with schema

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// Pattern 1: Evolve with registry (specific reader schema)
const registry = new SchemaRegistry({ host: '...' });
 
// Consumer specifies reader schema (its view of the data)
const readerSchema = await registry.getLatestSchemaForSubject('order-created-value');
const deserializer = new AvroDeserializer(registry, {
  readerSchema: readerSchema,  // Use latest schema we understand
});
 
// Avro resolves differences between writer and reader schemas
// - Extra fields in writer: ignored
// - Missing fields in writer: use defaults from reader
// - Type promotions: automatic (int -> long)
 
// Pattern 2: Multiple schema versions support
class OrderConsumer {
  private schemaHandlers = new Map<number, (data: any) => Order>();
  
  async initialize() {
    // Fetch all known schema versions
    const versions = await registry.getAllVersions('order-created-value');
    
    for (const version of versions) {
      const schema = await registry.getSchemaByVersion('order-created-value', version);
      this.schemaHandlers.set(
        schema.id,
        this.createHandler(schema)
      );
    }
  }
  
  async process(message: Buffer): Promise<Order> {
    const schemaId = readSchemaId(message);  // Read from magic bytes
    const handler = this.schemaHandlers.get(schemaId);
    
    if (!handler) {
      // Unknown schema - fetch and add dynamically
      const schema = await registry.getSchema(schemaId);
      this.schemaHandlers.set(schemaId, this.createHandler(schema));
    }
    
    const rawData = deserialize(message, schema);
    return handler(rawData);
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

// Compatibility check flow
 
// 1. Developer proposes new schema version
const newSchema = `{
  "type": "record",
  "name": "OrderCreated",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "totalAmount", "type": "double"},
    {"name": "currency", "type": "string"}  // NEW: Added without default!
  ]
}`;
 
// 2. Schema registry checks against existing versions
const checkResult = await registry.testCompatibility(
  'order-created-value',
  newSchema
);
 
// 3. Registry returns compatibility result
console.log(checkResult);
// {
//   isCompatible: false,
//   errors: [
//     {
//       errorType: 'READER_FIELD_MISSING_DEFAULT_VALUE',
//       message: 'Field "currency" has no default value'
//     }
//   ]
// }
 
// 4. Registration rejected
try {
  await registry.register('order-created-value', newSchema);
} catch (error) {
  // SchemaRegistryError: Schema is not compatible
}
 
// 5. Developer fixes schema (add default)
const fixedSchema = `{
  "type": "record",
  "name": "OrderCreated",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "totalAmount", "type": "double"},
    {"name": "currency", "type": "string", "default": "USD"}  // FIXED!
  ]
}`;
 
// 6. New check passes
const fixedResult = await registry.testCompatibility(
  'order-created-value',
  fixedSchema
);
// { isCompatible: true, errors: [] }
 
// 7. Registration succeeds
await registry.register('order-created-value', fixedSchema);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

name: Schema Compatibility Check
 
on:
  pull_request:
    paths:
      - 'schemas/**'
      - 'src/main/avro/**'
 
jobs:
  check-compatibility:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Check Schema Compatibility
        run: |
          # Check each modified schema against registry
          for schema_file in $(git diff --name-only origin/main -- 'schemas/*.avsc'); do
            subject=$(basename $schema_file .avsc)-value
            
            echo "Checking $subject..."
            
            response=$(curl -s -X POST \
              -H "Content-Type: application/vnd.schemaregistry.v1+json" \
              --data-binary "@$schema_file" \
              "$SCHEMA_REGISTRY_URL/compatibility/subjects/$subject/versions/latest")
            
            if [ "$(echo $response | jq .is_compatible)" != "true" ]; then
              echo "❌ Schema $subject is NOT compatible!"
              echo "$response" | jq .
              exit 1
            fi
            
            echo "✅ Schema $subject is compatible"
          done
          
      - name: Register Schemas (on merge)
        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
        run: |
          for schema_file in schemas/*.avsc; do
            subject=$(basename $schema_file .avsc)-value
            curl -X POST \
              -H "Content-Type: application/vnd.schemaregistry.v1+json" \
              --data-binary "@$schema_file" \
              "$SCHEMA_REGISTRY_URL/subjects/$subject/versions"
          done

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

// Schema metadata for governance
interface SchemaMetadata {
  // Ownership
  owner: {
    team: string;              // "commerce-team"
    contact: string;           // "commerce@company.com"
    slackChannel: string;      // "#commerce-platform"
  };
  
  // Lifecycle
  lifecycle: {
    status: 'DRAFT' | 'ACTIVE' | 'DEPRECATED' | 'SUNSET';
    createdAt: Date;
    deprecatedAt?: Date;
    sunsetDate?: Date;
    successor?: string;        // New schema replacing this one
  };
  
  // Documentation
  documentation: {
    description: string;
    usageNotes: string;
    samplePayload: string;
    changeLog: ChangeLogEntry[];
  };
  
  // Compliance
  compliance: {
    containsPII: boolean;
    dataClassification: 'PUBLIC' | 'INTERNAL' | 'CONFIDENTIAL' | 'RESTRICTED';
    retentionPolicy: string;
    gdprRelevant: boolean;
  };
  
  // Dependencies
  consumers: ConsumerInfo[];   // Known consumers of this schema
  dependencies: string[];      // Other schemas this schema references
}
 
// Apicurio-style metadata rules
const schemaRules = {
  // Require description
  {
    type: 'VALIDITY',
    config: {
      requireDescription: true,
      minDescriptionLength: 50,
    },
  },
  // Require owner metadata
  {
    type: 'METADATA',
    config: {
      requiredLabels: ['owner-team', 'data-classification'],
    },
  },
  // Naming convention
  {
    type: 'NAMING',
    config: {
      pattern: '^[a-z]+-[a-z]+(-[a-z]+)*$',  // kebab-case
      message: 'Subject names must be kebab-case',
    },
  },
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

// Best practices for registry performance
 
// 1. Client-side caching with TTL
const schemaCache = new LRUCache<number, Schema>({
  max: 1000,           // Max schemas to cache
  ttl: 1000 * 60 * 60, // 1 hour TTL (schemas rarely change)
});
 
async function getSchema(schemaId: number): Promise<Schema> {
  const cached = schemaCache.get(schemaId);
  if (cached) return cached;
  
  const schema = await registry.getSchema(schemaId);
  schemaCache.set(schemaId, schema);
  return schema;
}
 
// 2. Warm cache on startup
async function warmSchemaCache(subjects: string[]) {
  console.log('Warming schema cache...');
  
  for (const subject of subjects) {
    const versions = await registry.getAllVersions(subject);
    for (const version of versions) {
      const schema = await registry.getSchemaByVersion(subject, version);
      schemaCache.set(schema.id, schema);
    }
  }
  
  console.log(`Cached ${schemaCache.size} schemas`);
}
 
// 3. Fallback for registry unavailability
class ResilientSchemaClient {
  private localFallback: Map<number, Schema>;
  
  constructor() {
    // Load critical schemas from bundled backup
    this.localFallback = loadBundledSchemas();
  }
  
  async getSchema(schemaId: number): Promise<Schema> {
    try {
      return await this.fetchWithRetry(schemaId);
    } catch (error) {
      // Registry unreachable - use bundled fallback
      const fallback = this.localFallback.get(schemaId);
      if (fallback) {
        logger.warn(`Using bundled schema for ID ${schemaId}`);
        return fallback;
      }
      throw new Error(`Schema ${schemaId} unavailable; no fallback`);
    }
  }
}
 
// 4. Health check and alerting
async function checkRegistryHealth(): Promise<HealthStatus> {
  const start = Date.now();
  
  try {
    await registry.listSubjects();
    const latency = Date.now() - start;
    
    return {
      status: latency < 100 ? 'HEALTHY' : 'DEGRADED',
      latency,
    };
  } catch (error) {
    return { status: 'UNHEALTHY', error: error.message };
  }
}