附录：随书实验完整代码

services:
  mysql:
    image: mysql@sha256:0f51e63e537a1f176423fa03ef559aabe6fa4372a53160a44bc7e75da7d15364
    command: [--server-id=184054, --log-bin=mysql-bin, --binlog-format=ROW, --binlog-row-image=FULL, --binlog-expire-logs-seconds=86400]
    environment:
      MYSQL_ROOT_PASSWORD: cdc-book-local
      MYSQL_ROOT_HOST: '%'
    ports: ['127.0.0.1:23316:3306']
  kafka:
    image: apache/kafka@sha256:4ceccc577f03f51f6af8dbfda55194d0d892f4fa7913ffbded567ce3895622ed
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,HOST://:29092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,HOST://127.0.0.1:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    ports: ['127.0.0.1:29092:29092']
  connect:
    image: quay.io/debezium/connect@sha256:1290a7abb0c1fe957f8d732e43efbea24201d9ba0fa5ba79582f76438c05d0ba
    depends_on: [mysql, kafka]
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: cdc-book-connect
      CONFIG_STORAGE_TOPIC: book_connect_configs
      OFFSET_STORAGE_TOPIC: book_connect_offsets
      STATUS_STORAGE_TOPIC: book_connect_status
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      KEY_CONVERTER_SCHEMAS_ENABLE: 'false'
      VALUE_CONVERTER_SCHEMAS_ENABLE: 'false'
      REST_ADVERTISED_HOST_NAME: connect
    ports: ['127.0.0.1:28083:8083']
  postgres:
    image: postgres@sha256:a63c1ea3f6a3a5a1621f97d2aba3b30faecfab4224c1b0000cc85ced57bb1939
    environment:
      POSTGRES_USER: book
      POSTGRES_PASSWORD: cdc-book-local
      POSTGRES_DB: book
    ports: ['127.0.0.1:25436:5432']

#!/usr/bin/env python3
"""Real Kafka consumer; database transaction commits before broker offset commit."""
import argparse
import json
import os
import signal
import time
from pathlib import Path

import psycopg
from confluent_kafka import Consumer, KafkaError

DSN = 'postgresql://book:cdc-book-local@127.0.0.1:25436/book'


def apply(db, group, generation, message):
    value = json.loads(message.value()) if message.value() is not None else None
    key = json.loads(message.key()) if message.key() else None
    # Kafka tombstones are compaction markers; the preceding Debezium delete
    # envelope, not the null value, carries the source position and operation.
    with db.transaction():
        fresh = db.execute('INSERT INTO delivery_history VALUES (%s,%s,%s,%s) ON CONFLICT DO NOTHING RETURNING 1',
                           (group, message.topic(), message.partition(), message.offset())).fetchone()
        if fresh and value:
            assert value['op'] in ('r', 'c', 'u', 'd'), value
            after = value['after']
            row = after if after is not None else value['before']
            assert key['id'] == row['id']
            document = {k: row[k] for k in ('id', 'sku', 'price_cents', 'active', 'revision')}
            if 'note' in row:
                document['note'] = row['note']
            # Same primary key is always on one partition; no cross-partition
            # numeric offset comparison is claimed here.
            db.execute('''INSERT INTO products_projection VALUES (%s,%s,%s,%s,%s,%s)
              ON CONFLICT(generation,id) DO UPDATE SET document=EXCLUDED.document,
              deleted=EXCLUDED.deleted,source_partition=EXCLUDED.source_partition,
              source_offset=EXCLUDED.source_offset
              WHERE products_projection.source_partition=EXCLUDED.source_partition
              AND products_projection.source_offset<EXCLUDED.source_offset''',
              (generation, row['id'], psycopg.types.json.Jsonb(document), value['op'] == 'd', message.partition(), message.offset()))
        db.execute('''INSERT INTO delivery_checkpoint VALUES (%s,%s,%s,%s)
            ON CONFLICT(consumer_group,topic,partition) DO UPDATE
            SET next_offset=GREATEST(delivery_checkpoint.next_offset,EXCLUDED.next_offset)''',
            (group, message.topic(), message.partition(), message.offset()+1))
    return bool(fresh), value


def bind(db, generation, group, topic):
    with db.transaction():
        db.execute('INSERT INTO consumer_binding VALUES (%s,%s,%s) ON CONFLICT DO NOTHING', (generation,group,topic))
        actual = db.execute('SELECT consumer_group,topic FROM consumer_binding WHERE generation=%s', (generation,)).fetchone()
        if actual != (group,topic):
            raise ValueError('Generation must bind to one unique consumer group and topic')


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--group', required=True)
    parser.add_argument('--generation', required=True)
    parser.add_argument('--topic', required=True)
    parser.add_argument('--crash-after-db', action='store_true')
    parser.add_argument('--ready-file')
    parser.add_argument('--replay', action='store_true')
    parser.add_argument('--idle-seconds', type=float, default=3)
    args = parser.parse_args()
    db = psycopg.connect(DSN, autocommit=True)
    bind(db, args.generation, args.group, args.topic)
    consumer = Consumer({'bootstrap.servers': '127.0.0.1:29092', 'group.id': args.group,
                         'enable.auto.commit': False, 'auto.offset.reset': 'earliest',
                         'enable.auto.offset.store': False})
    def assigned(c, partitions):
        if args.replay:
            for partition in partitions:
                partition.offset = 0
        c.assign(partitions)
    consumer.subscribe([args.topic], on_assign=assigned)
    totals = {'messages': 0, 'fresh': 0, 'duplicates': 0, 'ops': {}, 'source_positions': []}
    last = time.monotonic()
    try:
        while True:
            msg = consumer.poll(0.5)
            if msg is None:
                if totals['messages'] and time.monotonic()-last > args.idle_seconds:
                    break
                if time.monotonic()-last > 60:
                    raise TimeoutError('No messages available; integration may not be skipped')
                continue
            if msg.error():
                raise RuntimeError(str(msg.error()))
            fresh, value = apply(db, args.group, args.generation, msg)
            totals['messages'] += 1
            totals['fresh' if fresh else 'duplicates'] += 1
            if value:
                totals['ops'][value['op']] = totals['ops'].get(value['op'], 0)+1
                if len(totals['source_positions']) < 8:
                    totals['source_positions'].append(value['source'])
            if args.crash_after_db:
                if args.ready_file:
                    Path(args.ready_file).write_text(json.dumps({'topic': msg.topic(), 'partition': msg.partition(), 'offset': msg.offset()}))
                print('database committed; waiting for controller SIGKILL', flush=True)
                while True:
                    time.sleep(1)
            consumer.commit(message=msg, asynchronous=False)
            last = time.monotonic()
    finally:
        consumer.close()
        db.close()
    print(json.dumps(totals, ensure_ascii=False), flush=True)


if __name__ == '__main__':
    main()

CREATE TABLE products_projection (
  generation TEXT NOT NULL,
  id BIGINT NOT NULL,
  document JSONB,
  deleted BOOLEAN NOT NULL,
  source_partition INTEGER NOT NULL,
  source_offset BIGINT NOT NULL,
  PRIMARY KEY(generation,id)
);
CREATE TABLE delivery_checkpoint (
  consumer_group TEXT NOT NULL,
  topic TEXT NOT NULL,
  partition INTEGER NOT NULL,
  next_offset BIGINT NOT NULL,
  PRIMARY KEY(consumer_group,topic,partition)
);
CREATE TABLE delivery_history (
  consumer_group TEXT NOT NULL,
  topic TEXT NOT NULL,
  partition INTEGER NOT NULL,
  offset_value BIGINT NOT NULL,
  PRIMARY KEY(consumer_group,topic,partition,offset_value)
);
CREATE TABLE consumer_binding (
  generation TEXT PRIMARY KEY,
  consumer_group TEXT NOT NULL UNIQUE,
  topic TEXT NOT NULL
);

#!/usr/bin/env python3
"""Fail-closed integration against real Docker MySQL/Debezium/Kafka/Postgres."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import signal
import subprocess
import sys
import time
import urllib.request

import pymysql
import psycopg
from confluent_kafka import Consumer, TopicPartition
from confluent_kafka.admin import AdminClient

ROOT = Path(__file__).resolve().parents[1]
LAB = ROOT/'lab'
EVIDENCE = Path(os.environ.get('CDC_EVIDENCE_DIR', str(LAB/'evidence')))
EVIDENCE.mkdir(parents=True, exist_ok=True)
checks = []


def check(name, condition, details=None):
    record = {'name': name, 'passed': bool(condition), 'details': details}
    checks.append(record)
    (EVIDENCE/'checks.json').write_text(json.dumps(checks, ensure_ascii=False, indent=2))
    if not condition:
        raise AssertionError(record)
    print('PASS', name, flush=True)


def eventually(fn, seconds=180):
    deadline = time.monotonic()+seconds
    last = None
    while time.monotonic() < deadline:
        try:
            result = fn()
            if result:
                return result
        except Exception as exc:
            last = str(exc)
        time.sleep(1)
    raise TimeoutError(f'Integration dependency did not become ready: {last}')


def mysql():
    return pymysql.connect(host='127.0.0.1', port=23316, user='root', password='cdc-book-local', autocommit=True,
                           cursorclass=pymysql.cursors.DictCursor)


def sql(connection, query, args=None):
    with connection.cursor() as cursor:
        cursor.execute(query, args)
        return cursor.fetchall()


def request(path, payload=None):
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request('http://127.0.0.1:28083'+path, data=data,
                                 headers={'Content-Type': 'application/json'})
    with urllib.request.urlopen(req, timeout=15) as response:
        return json.load(response)


def connector(name, prefix, slow=False):
    cfg = {'connector.class': 'io.debezium.connector.mysql.MySqlConnector', 'tasks.max': '1',
      'key.converter': 'org.apache.kafka.connect.json.JsonConverter', 'key.converter.schemas.enable': 'false',
      'value.converter': 'org.apache.kafka.connect.json.JsonConverter', 'value.converter.schemas.enable': 'false',
      'database.hostname': 'mysql', 'database.port': '3306', 'database.user': 'cdc',
      'database.password': 'cdc-book-local', 'database.server.id': '184055' if slow else '184056',
      'topic.prefix': prefix, 'database.include.list': 'inventory', 'table.include.list': 'inventory.products',
      'schema.history.internal.kafka.bootstrap.servers': 'kafka:9092',
      'schema.history.internal.kafka.topic': prefix+'.schema-history', 'snapshot.mode': 'initial',
      'include.schema.changes': 'false', 'topic.creation.default.replication.factor': '1',
      'topic.creation.default.partitions': '1', 'heartbeat.interval.ms': '1000'}
    if slow:
        cfg['snapshot.select.statement.overrides'] = 'inventory.products'
        cfg['snapshot.select.statement.overrides.inventory.products'] = 'SELECT * FROM inventory.products WHERE SLEEP(0.003)=0'
    request('/connectors', {'name': name, 'config': cfg})
    return cfg


def worker(group, generation, topic, extra=()):
    admin = AdminClient({'bootstrap.servers':'127.0.0.1:29092'})
    eventually(lambda: topic in admin.list_topics(timeout=10).topics)
    command = [sys.executable, str(LAB/'consumer.py'), '--group', group, '--generation', generation,
               '--topic', topic, *extra]
    result = subprocess.run(command, capture_output=True, text=True, timeout=150)
    (EVIDENCE/(group+('-replay' if '--replay' in extra else '')+'.log')).write_text(result.stdout+'\n'+result.stderr)
    if result.returncode:
        raise RuntimeError(f'Consumer failed ({group}): {result.stdout}\n{result.stderr}')
    return json.loads(result.stdout.strip().splitlines()[-1])


def source_rows(source):
    rows = sql(source, 'SELECT * FROM inventory.products ORDER BY id')
    return rows


def projection_rows(db, generation):
    return [row[0] for row in db.execute('SELECT document FROM products_projection WHERE generation=%s AND NOT deleted ORDER BY id', (generation,)).fetchall()]


def reconcile(source, db, generation, label):
    a = source_rows(source)
    b = projection_rows(db, generation)
    dump = lambda rows: json.dumps(rows, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode()
    evidence = {'source_rows': len(a), 'projection_rows': len(b), 'source_sha256': hashlib.sha256(dump(a)).hexdigest(),
                'projection_sha256': hashlib.sha256(dump(b)).hexdigest(), 'all_rows_all_fields_equal': a == b}
    (EVIDENCE/(label+'-reconciliation.json')).write_text(json.dumps(evidence, indent=2))
    check(label+' exact full-row reconciliation', a == b, evidence)
    return evidence


def main():
    start = time.time()
    source = eventually(mysql)
    pg = eventually(lambda: psycopg.connect('postgresql://book:cdc-book-local@127.0.0.1:25436/book', autocommit=True))
    eventually(lambda: request('/connector-plugins'))
    versions = {'mysql': sql(source, 'SELECT VERSION() AS version')[0]['version'],
                'postgres': pg.execute('SELECT version()').fetchone()[0],
                'connect_plugins': request('/connector-plugins'), 'python': sys.version,
                'kafka': (EVIDENCE/'kafka-version.txt').read_text().strip(),
                'images': {name:json.loads((EVIDENCE/('image-'+name+'.json')).read_text()) for name in ('mysql','kafka','connect','postgres')}}
    (EVIDENCE/'versions.json').write_text(json.dumps(versions, indent=2))
    for statement in (LAB/'schema.sql').read_text().split(';'):
        if statement.strip():
            sql(source, statement)
    with source.cursor() as cursor:
        cursor.executemany('INSERT INTO inventory.products VALUES (%s,%s,%s,%s,%s)',
                           [(i, 'sku-'+str(i), i*17, 1, 1) for i in range(1,2001)])
    pg.execute((LAB/'projection.sql').read_text())
    cfg = connector('book-source', 'book', slow=True)
    (EVIDENCE/'connector-config.json').write_text(json.dumps(cfg, indent=2))
    def active_scan():
        return [r for r in sql(source, 'SHOW FULL PROCESSLIST') if r.get('Info') and 'SELECT * FROM inventory.products WHERE SLEEP' in r['Info']]
    before = eventually(active_scan)
    timestamp_before = sql(source, 'SELECT UTC_TIMESTAMP(6) AS t')[0]['t'].isoformat()
    sql(source, "INSERT INTO inventory.products VALUES (3001,'snapshot-concurrent-insert',321,1,1)")
    sql(source, "UPDATE inventory.products SET price_cents=777,revision=2 WHERE id=1")
    sql(source, 'DELETE FROM inventory.products WHERE id=2')
    timestamp_after = sql(source, 'SELECT UTC_TIMESTAMP(6) AS t')[0]['t'].isoformat()
    after = active_scan()
    overlap = {'before': before, 'after': after, 'write_start_mysql_utc': timestamp_before,
               'write_end_mysql_utc': timestamp_after, 'write_ids': [3001,1,2]}
    (EVIDENCE/'snapshot-write-overlap.json').write_text(json.dumps(overlap, default=str, indent=2))
    check('writes occurred while real snapshot SELECT remained active', bool({r['Id'] for r in before} & {r['Id'] for r in after}), overlap)
    first = worker('main', 'live', 'book.inventory.products')
    (EVIDENCE/'initial-consume.json').write_text(json.dumps(first, indent=2))
    check('real snapshot and binlog r/c/u/d captured', all(first['ops'].get(op,0)>0 for op in ('r','c','u','d')), first['ops'])
    check('real binlog file and position captured', any(p.get('file') and p.get('pos') for p in first['source_positions']))
    reconcile(source, pg, 'live', 'snapshot-plus-binlog')
    replay = worker('main', 'live', 'book.inventory.products', ['--replay'])
    check('duplicate replay detected in transactional inbox', replay['duplicates'] == first['messages'] and replay['fresh'] == 0, replay)
    reconcile(source, pg, 'live', 'duplicate-replay')
    before_invalid = (pg.execute('SELECT count(*) FROM products_projection').fetchone()[0], pg.execute('SELECT count(*) FROM delivery_history').fetchone()[0])
    rejections=[]
    for group,generation,topic in [('main','forbidden-new','book.inventory.products'),('other','live','book.inventory.products'),('main','live','wrong-topic')]:
        bad=subprocess.run([sys.executable,str(LAB/'consumer.py'),'--group',group,'--generation',generation,'--topic',topic],capture_output=True,text=True,timeout=15)
        rejections.append({'group':group,'generation':generation,'topic':topic,'returncode':bad.returncode,'explicit_binding_error':'Generation must bind' in bad.stderr})
    after_invalid = (pg.execute('SELECT count(*) FROM products_projection').fetchone()[0], pg.execute('SELECT count(*) FROM delivery_history').fetchone()[0])
    check('generation group topic mismatch rejected without pollution', all(r['returncode']!=0 and r['explicit_binding_error'] for r in rejections) and before_invalid==after_invalid and pg.execute('SELECT count(*) FROM consumer_binding').fetchone()[0]==1, rejections)
    # Fresh group: one durable DB transaction, then forcibly kill the real process
    # while its code is blocked before consumer.commit().
    marker = EVIDENCE/'kill-ready.json'
    command = [sys.executable, str(LAB/'consumer.py'), '--group', 'crash', '--generation', 'crash-gen',
               '--topic', 'book.inventory.products', '--crash-after-db', '--ready-file', str(marker)]
    with (EVIDENCE/'crashed-process.log').open('w') as log:
        child = subprocess.Popen(command, stdout=log, stderr=log)
        eventually(lambda: marker.exists(), 60)
        position = json.loads(marker.read_text())
        check('database committed before process kill', pg.execute('SELECT count(*) FROM delivery_history WHERE consumer_group=%s', ('crash',)).fetchone()[0] == 1)
        probe = Consumer({'bootstrap.servers':'127.0.0.1:29092','group.id':'crash','enable.auto.commit':False})
        committed = probe.committed([TopicPartition(position['topic'],position['partition'])], timeout=10)[0].offset
        probe.close()
        check('broker offset uncommitted at kill barrier', committed < 0 or committed <= position['offset'], {'broker_next_offset':committed, **position})
        child.kill()
        child.wait(timeout=10)
        check('real subprocess terminated by SIGKILL', child.returncode == -signal.SIGKILL)
    restarted = worker('crash', 'crash-gen', 'book.inventory.products')
    check('restart re-delivers committed-database event', restarted['duplicates'] >= 1, restarted)
    reconcile(source, pg, 'crash-gen', 'crash-restart')
    # No consumer runs while these real source transactions create backlog.
    for i in range(100):
        sql(source, 'UPDATE inventory.products SET price_cents=price_cents+1,revision=revision+1 WHERE id=%s', (10+i,))
    sql(source, 'DELETE FROM inventory.products WHERE id=10')
    sql(source, "INSERT INTO inventory.products VALUES (4001,'after-outage',222,1,1)")
    stale = source_rows(source) != projection_rows(pg, 'live')
    check('stopped consumer leaves projection observably stale', stale)
    caught = worker('main', 'live', 'book.inventory.products')
    check('backlog drained after consumer restart', caught['ops'].get('u',0)>=100 and caught['ops'].get('d',0)>=1, caught)
    reconcile(source, pg, 'live', 'backlog-catchup')
    sql(source, "ALTER TABLE inventory.products ADD COLUMN note VARCHAR(80) NULL")
    sql(source, "UPDATE inventory.products SET note='new schema field',revision=revision+1 WHERE id=1")
    # Existing untouched rows receive the schema default on rebuild; live schema
    # migration normalizes the same null default before reading new events.
    pg.execute("UPDATE products_projection SET document=document || '{\"note\":null}'::jsonb WHERE generation='live'")
    evolved = worker('main', 'live', 'book.inventory.products')
    check('real ALTER and subsequent binlog update consumed', evolved['ops'].get('u',0)>=1)
    check('added field reaches query document', pg.execute("SELECT document->>'note' FROM products_projection WHERE generation='live' AND id=1").fetchone()[0] == 'new schema field')
    reconcile(source, pg, 'live', 'schema-evolution')
    connector('book-rebuild', 'rebuild', slow=False)
    rebuilt = worker('rebuild', 'new-generation', 'rebuild.inventory.products')
    check('isolated new snapshot rebuild consumed', rebuilt['ops'].get('r',0)==len(source_rows(source)), rebuilt['ops'])
    reconcile(source, pg, 'new-generation', 'full-rebuild')
    check('live generation preserved during rebuild', projection_rows(pg,'live') == projection_rows(pg,'new-generation'))
    statuses = {name: request('/connectors/'+name+'/status') for name in ('book-source','book-rebuild')}
    check('both actual connectors and tasks RUNNING', all(s['connector']['state']=='RUNNING' and len(s['tasks'])==1 and all(t['state']=='RUNNING' for t in s['tasks']) for s in statuses.values()), statuses)
    result = {'status':'PASS','skipped':0,'checks':checks,'check_count':len(checks),'duration_seconds':round(time.time()-start,3),
              'versions':versions,'run_url':os.environ.get('BOOK_RUN_URL'),'source_revision':os.environ.get('BOOK_SOURCE_SHA'),
              'boundaries':['single Kafka broker; one data partition','real Debezium initial snapshot and MySQL binlog',
                            'local teaching credentials only','no multi-node failover or binlog-loss proof',
                            'cross-store delivery is at-least-once with projection convergence; not end-to-end exactly-once']}
    (EVIDENCE/'runtime.json').write_text(json.dumps(result, ensure_ascii=False, indent=2))
    print(json.dumps({'status':'PASS','checks':len(checks),'seconds':result['duration_seconds']}))


if __name__ == '__main__':
    main()

#!/bin/sh
set -eu
cd "$(dirname "$0")/.."
: "${CDC_COMPOSE_PROJECT:?Set a dedicated CDC_COMPOSE_PROJECT before running; this lab owns only the dedicated project resources}"
case "$CDC_COMPOSE_PROJECT" in cdc-book-*) ;; *) echo 'Project must start cdc-book-' >&2; exit 2;; esac
mkdir -p "${CDC_EVIDENCE_DIR:-lab/evidence}"
if [ -f "${CDC_EVIDENCE_DIR:-lab/evidence}/kill-ready.json" ]; then
  echo "Refusing stale crash marker; use a fresh evidence directory" >&2
  exit 2
fi
cleanup() {
  docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml logs --no-color > "${CDC_EVIDENCE_DIR:-lab/evidence}/containers.log" 2>&1 || true
  docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml down --volumes --remove-orphans
}
trap cleanup EXIT
python3 -m pip install -r lab/requirements.txt
docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml up -d
for task_service in mysql kafka connect postgres; do
  task_container=$(docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml ps -q "$task_service")
  task_image=$(docker inspect --format '{{.Image}}' "$task_container")
  docker image inspect --format '{{json .RepoDigests}}' "$task_image" > "${CDC_EVIDENCE_DIR:-lab/evidence}/image-$task_service.json"
done
docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml exec -T kafka /opt/kafka/bin/kafka-topics.sh --version > "${CDC_EVIDENCE_DIR:-lab/evidence}/kafka-version.txt"
python3 lab/run.py

CREATE DATABASE inventory;
CREATE USER 'cdc'@'%' IDENTIFIED BY 'cdc-book-local';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, LOCK TABLES ON *.* TO 'cdc'@'%';
CREATE TABLE inventory.products (
  id BIGINT PRIMARY KEY,
  sku VARCHAR(64) NOT NULL,
  price_cents INT NOT NULL,
  active TINYINT NOT NULL,
  revision INT NOT NULL
);