《Java 消息中间件·事件驱动架构》6.3 消费者驱动的契约

// ConsumerContractTest.java — 消费者驱动的契约纯 Java 断言

import java.util.*;

public class ConsumerContractTest {

    record SchemaField(String name, String type, boolean required, String description) {}

    static class Contract {
        final String consumerName;
        final Set<SchemaField> requiredFields;
        final Set<String> forbiddenFields;
        final Map<String, String> typeExpectations;

        Contract(String consumerName,
                 Set<SchemaField> requiredFields,
                 Set<String> forbiddenFields,
                 Map<String, String> typeExpectations) {
            this.consumerName = consumerName;
            this.requiredFields = requiredFields;
            this.forbiddenFields = forbiddenFields;
            this.typeExpectations = typeExpectations;
        }

        public static Builder builder(String consumerName) {
            return new Builder(consumerName);
        }

        static class Builder {
            private final String name;
            private final Set<SchemaField> req = new LinkedHashSet<>();
            private final Set<String> forb = new HashSet<>();
            private final Map<String, String> types = new LinkedHashMap<>();

            Builder(String name) { this.name = name; }
            Builder requires(String name, String type, String desc) {
                req.add(new SchemaField(name, type, true, desc));
                types.put(name, type);
                return this;
            }
            Builder optional(String name, String type, String desc) {
                req.add(new SchemaField(name, type, false, desc));
                types.put(name, type);
                return this;
            }
            Builder forbids(String name) {
                forb.add(name);
                return this;
            }
            Contract build() { return new Contract(name, req, forb, types); }
        }
    }

    record Result(boolean pass, List<String> failures) {}

    static Result verify(Map<String, String> producerSchema, Contract contract) {
        List<String> failures = new ArrayList<>();

        for (SchemaField f : contract.requiredFields) {
            if (f.required && !producerSchema.containsKey(f.name())) {
                failures.add("[MISSING] required field '" + f.name() + "' ("
                    + f.type() + ") — " + f.description());
            }
        }

        for (Map.Entry<String, String> exp : contract.typeExpectations.entrySet()) {
            String name = exp.getKey();
            String expected = exp.getValue();
            if (producerSchema.containsKey(name)
                    && !producerSchema.get(name).equals(expected)) {
                failures.add("[TYPE_MISMATCH] field '" + name + "': expected "
                    + expected + ", got " + producerSchema.get(name));
            }
        }

        for (String f : contract.forbiddenFields) {
            if (producerSchema.containsKey(f)) {
                failures.add("[FORBIDDEN] field '" + f
                    + "' should not be present; it was removed/renamed");
            }
        }

        return new Result(failures.isEmpty(), failures);
    }

    static Contract projectionConsumer() {
        return Contract.builder("catalog-projection")
            .requires("loanId", "string", "主键，投影表行标识")
            .requires("bookId", "string", "图书关联，分组/过滤用")
            .requires("borrowedAt", "timestamp", "借出时间，列表页排序")
            .requires("dueAt", "timestamp", "到期时间，逾期提醒计算")
            .optional("occurredAt", "timestamp", "事件时间，排序备用")
            .forbids("borrowedAtOld")
            .build();
    }

    public static void main(String[] args) {
        // ... 四个场景见下方实测输出 ...
    }
}

=== 场景 A：生产者 v1.0.0，契约通过 ===
契约方: catalog-projection
producer fields: [loanId, bookId, memberId, borrowedAt, dueAt, occurredAt]
required fields: [loanId, bookId, borrowedAt, dueAt]
RESULT: PASS (failures=0)

=== 场景 B：生产者 v1.2.0，borrowedAt → startAt，契约红了 ===
契约方: catalog-projection
producer fields: [loanId, bookId, memberId, startAt, dueAt, occurredAt]
required fields: [loanId, bookId, borrowedAt, dueAt]
RESULT: FAIL (failures=1)
  [MISSING] required field 'borrowedAt' (timestamp) — 借出时间，列表页排序

=== 场景 C：生产者 v1.1.0，加可选字段 reason，契约不受影响 ===
producer fields: [loanId, bookId, memberId, borrowedAt, dueAt, occurredAt, reason]
RESULT: PASS (failures=0)

=== 场景 D：borrowedAt timestamp→long，契约红了 ===
producer fields: [loanId, bookId, memberId, borrowedAt, dueAt, occurredAt]
RESULT: FAIL (failures=1)
  [TYPE_MISMATCH] field 'borrowedAt': expected timestamp, got long

public class ContractVerification {
    static final List<Contract> ALL_CONTRACTS = List.of(
        ConsumerContractTest.projectionConsumer(),
        // notificationConsumer(), // 通知服务
        // overdueConsumer()       // 逾期调度
    );

    public static void main(String[] args) {
        Map<String, String> currentSchema = loadCurrentSchemaFromCatalog();
        boolean allPass = true;
        for (Contract c : ALL_CONTRACTS) {
            Result r = ConsumerContractTest.verify(currentSchema, c);
            System.out.println(c.consumerName + ": " + (r.pass() ? "PASS" : "FAIL"));
            for (String f : r.failures()) System.out.println("  " + f);
            if (!r.pass()) allPass = false;
        }
        if (!allPass) {
            System.exit(1); // mvn verify 看到非零退出码即失败
        }
    }

    static Map<String, String> loadCurrentSchemaFromCatalog() {
        // 从 events.yaml 读取当前版本的 schema，或从编译后的 record 反射导出
        // 2.3 的 CatalogSchemaDump 已经提供了反射导出逻辑，直接复用
        return Map.of(/* ... */);
    }
}

{
  "loanId": "L-001",
  "borrowedAt": "2026-10-20T09:00:00Z",
  "loanStartEpoch": 1729414800000
}