#!/usr/bin/env python3 """U48 公开入口：Spring AMQP（Qpid 替身 broker）与 Spring Kafka（真实 KRaft 4.1.2）集成实验。 - AMQP：Qpid Broker-J 内存 broker 以独立进程启动（AMQP 0-9-1 与 rabbit 客户端兼容，如实标注替身）。 - Kafka：真实 Kafka 4.1.2 KRaft 单机（Maven 构件组装 classpath），runner 先格式化存储、再启动 broker、 运行 Spring Kafka 实验、最后终止并清理临时目录。 - Pulsar：真实 broker 需下载官方二进制，未授权时如实受阻（不伪装）。 """ import hashlib import json import os from pathlib import Path import platform import re import shutil import signal import subprocess import sys import tempfile import time import uuid from datetime import datetime ROOT = Path(__file__).resolve().parent EVIDENCE = ROOT / 'evidence/u48-2026-10-31.json' SOURCES = ['pom.xml', 'README.md', 'run.py', 'qpid-config.json', 'src/main/java/study/u48/AmqpLab.java', 'src/main/java/study/u48/KafkaLab.java'] AMQP_PORT = 29991 KAFKA_PORT = 29992 KAFKA_CONTROLLER_PORT = 29993 REQUIRED_JARS = ['spring-rabbit-4.0.6.jar', 'spring-kafka-4.0.8.jar', 'amqp-client-5.27.1.jar', 'kafka-clients-4.1.2.jar', 'kafka_2.13-4.1.2.jar', 'qpid-broker-core-10.1.1.jar'] AMQP_MARKERS = ['amqpBroker ', 'rabbitDeclared ', 'rabbitLifecycle ', 'rabbitDlq ', 'rabbitRetry ', 'cleanup '] KAFKA_MARKERS = ['kafkaListener ', 'kafkaLifecycle ', 'kafkaManualAck ', 'kafkaErrorHandler ', 'kafkaTransaction ', 'cleanup '] def now(): return datetime.now().astimezone().isoformat() def diagnostics(output): if isinstance(output, bytes): output = output.decode('utf-8', 'replace') lines = [line for line in (output or '').splitlines() if any(x in line for x in ('Exception', 'ERROR', 'Caused by:', 'AssertionError'))] return [line[:400] for line in lines[-8:]] def reachable(port): import socket try: with socket.create_connection(('127.0.0.1', port), timeout=2): return True except OSError: return False def start_process(args, env, log_path): log = log_path.open('w', encoding='utf-8') process = subprocess.Popen(args, cwd=ROOT, env=env, text=True, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) return process, log def stop_process(process, log): if process and process.poll() is None: try: os.killpg(os.getpgid(process.pid), signal.SIGTERM) process.wait(timeout=15) except Exception: try: os.killpg(os.getpgid(process.pid), signal.SIGKILL) except Exception: pass if log: log.close() def wait_port(port, timeout_s, log_path): deadline = time.time() + timeout_s while time.time() < deadline: if reachable(port): return True time.sleep(0.5) raise RuntimeError(f'端口 {port} 未就绪；日志 {log_path}') def run_java(name, args, env, results, logs, timeout=180): log_path = logs / (name + '.log') started = now() run = subprocess.run(args, cwd=ROOT, env=env, capture_output=True, text=True, timeout=timeout) output = (run.stdout or '') + (run.stderr or '') result = {'name': name, 'exitCode': run.returncode, 'startedAt': started, 'finishedAt': now(), 'observed': [line for line in (run.stdout or '').splitlines() if line.startswith('U48 ')], 'log': str(log_path)} log_path.write_text(output, encoding='utf-8') results.append(result) if run.returncode != 0: print(name + ' exit=' + str(run.returncode) + '\n' + output[-3500:], file=sys.stderr, flush=True) raise RuntimeError(name + ' 失败') return result def main(): java_home = Path(os.environ.get('JAVA_HOME', '/Library/Java/JavaVirtualMachines/liberica-jdk-21.jdk/Contents/Home')) java = java_home / 'bin/java' mvn = shutil.which('mvn') if not java.is_file() or not mvn: raise RuntimeError('需要 JDK 21 和 Maven 3.9') allowed = ('PATH', 'HOME', 'TMPDIR', 'LANG', 'LC_ALL', 'TZ') env = {key: os.environ[key] for key in allowed if key in os.environ} env.update(JAVA_HOME=str(java_home), MAVEN_OPTS='-Xmx512m -XX:ActiveProcessorCount=2') java_version = subprocess.check_output([str(java), '-version'], stderr=subprocess.STDOUT, text=True, env=env).strip() if '21.0.11' not in java_version: raise RuntimeError('实验基线需要 JDK 21.0.11') result = {'issue': 85, 'input': 'python3 labs/broker-integrations-u48/run.py', 'startedAt': now(), 'finishedAt': None, 'environment': {'os': platform.platform(), 'arch': platform.machine(), 'java': java_version}, 'status': 'failed', 'exitCode': 1, 'phases': [], 'pulsar': None} run_id = uuid.uuid4().hex[:10] tmp_root = Path(tempfile.mkdtemp(prefix='u48-brokers-')) classpath = None qpid_proc = qpid_log = kafka_proc = kafka_log = None exit_code = 1 try: with tempfile.TemporaryDirectory(prefix='u47-classpath-') as directory: cp_file = Path(directory) / 'classpath.txt' subprocess.run([mvn, '-o', '-B', '-q', 'compile', 'org.apache.maven.plugins:maven-dependency-plugin:3.8.1:build-classpath', '-DincludeScope=runtime', f'-Dmdep.outputFile={cp_file}'], cwd=ROOT, env=env, check=True, timeout=300, capture_output=True, text=True) classpath = str(ROOT / 'target/classes') + os.pathsep + cp_file.read_text().strip() jars = [Path(x).name for x in cp_file.read_text().split(os.pathsep)] result['dependencies'] = [j for j in jars if any(k in j for k in ('spring-rabbit-4', 'spring-kafka-4', 'amqp-client-', 'kafka-clients-', 'kafka_2.13-', 'qpid-broker-core-', 'spring-pulsar-'))] for required in REQUIRED_JARS: if not any(required in j for j in result['dependencies']): raise RuntimeError('依赖版本不符: ' + required + '; actual=' + str(result['dependencies'])) # ---- 阶段 1：AMQP（Qpid 替身 broker） ---- qpid_config = tmp_root / 'qpid-config.json' qpid_config.write_text((ROOT / 'qpid-config.json').read_text(encoding='utf-8') .replace('${qpid.amqp_port}', str(AMQP_PORT)), encoding='utf-8') qpid_log_path = tmp_root / 'qpid-broker.log' vhost_config = '{"type":"Memory"}' # 自定义初始配置：Anonymous 提供者（rabbit 客户端侧注册 ANONYMOUS SASL）+ JSON 节点的 Memory 虚拟主机 logback_file = tmp_root / 'qpid-logback.xml' logback_file.write_text( '' '%d{HH:mm:ss.SSS} %-5level %logger{20} - %msg%n' '', encoding='utf-8') qpid_proc, qpid_log = start_process( [str(java), '-Xmx384m', f'-Dqpid.work_dir={tmp_root}/qpid-work', f'-Dlogback.configurationFile={logback_file}', '-cp', classpath, 'org.apache.qpid.server.Main', '-icp', str(qpid_config), '-sp', str(tmp_root / 'qpid-work'), '-st', 'JSON', '-prop', 'qpid.initial_config_virtualhost_config=' + vhost_config], env, qpid_log_path) wait_port(AMQP_PORT, 60, str(qpid_log_path)) run_java('amqp-lab', [str(java), '-Xms64m', '-Xmx384m', '-XX:ActiveProcessorCount=2', '-cp', classpath, 'study.u48.AmqpLab', str(AMQP_PORT)], env, result['phases'], tmp_root) # ---- 阶段 2：Kafka（真实 KRaft 单机） ---- kafka_data = tmp_root / 'kafka-data' kafka_data.mkdir() cluster_uuid = uuid.uuid4() server_props = tmp_root / 'kafka-server.properties' server_props.write_text( 'process.roles=broker,controller\n' 'node.id=1\n' f'controller.quorum.voters=1@127.0.0.1:{KAFKA_CONTROLLER_PORT}\n' f'listeners=PLAINTEXT://127.0.0.1:{KAFKA_PORT},CONTROLLER://127.0.0.1:{KAFKA_CONTROLLER_PORT}\n' f'advertised.listeners=PLAINTEXT://127.0.0.1:{KAFKA_PORT}\n' 'controller.listener.names=CONTROLLER\n' 'listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT\n' f'log.dirs={kafka_data}\n' 'num.partitions=1\n' 'offsets.topic.replication.factor=1\n' 'transaction.state.log.replication.factor=1\n' 'transaction.state.log.min.isr=1\n' 'group.initial.rebalance.delay.ms=0\n', encoding='utf-8') format_log = tmp_root / 'kafka-format.log' fmt = subprocess.run([str(java), '-Xmx256m', '-cp', classpath, 'kafka.tools.StorageTool', 'format', '-t', str(cluster_uuid), '-c', str(server_props), '--ignore-formatted'], env=env, capture_output=True, text=True, timeout=120) format_log.write_text((fmt.stdout or '') + (fmt.stderr or ''), encoding='utf-8') if fmt.returncode != 0: raise RuntimeError('Kafka 存储格式化失败；日志 ' + str(format_log)) kafka_log_path = tmp_root / 'kafka-broker.log' kafka_proc, kafka_log = start_process( [str(java), '-Xmx512m', '-XX:ActiveProcessorCount=2', '-cp', classpath, 'kafka.Kafka', str(server_props)], env, kafka_log_path) wait_port(KAFKA_PORT, 90, str(kafka_log_path)) time.sleep(2) run_java('kafka-lab', [str(java), '-Xms64m', '-Xmx384m', '-Du48.bootstrap=127.0.0.1:' + str(KAFKA_PORT), '-XX:ActiveProcessorCount=2', '-cp', classpath, 'study.u48.KafkaLab'], env, result['phases'], tmp_root) result['pulsar'] = { 'status': '受阻', 'reason': 'Spring Pulsar 2.0.8（BOM 钉版）验证需要真实 Pulsar broker；' '官方 Pulsar 发行包下载未获授权，本票如实受阻，不把固定源码推演当作运行。', } result['status'] = 'passed' result['exitCode'] = 0 result['finishedAt'] = now() result['limits'] = ('AMQP 用 Qpid Broker-J 内存替身（AMQP 0-9-1 兼容），非 RabbitMQ 本体；' 'Kafka 为真实 4.1.2 KRaft 单机；Pulsar 受阻待真实 broker；' '均未运行跨主机、集群与生产容量。') result['artifactSha256'] = {p: hashlib.sha256((ROOT / p).read_bytes()).hexdigest() for p in SOURCES if (ROOT / p).exists()} except BaseException as exc: result['error'] = type(exc).__name__ + ': ' + str(exc) result['diagnostics'] = diagnostics(str(exc)) result['finishedAt'] = now() print('FAILED: ' + str(exc), file=sys.stderr, flush=True) finally: stop_process(qpid_proc, qpid_log) stop_process(kafka_proc, kafka_log) if result['status'] == 'passed' and result['exitCode'] == 0: shutil.rmtree(tmp_root, ignore_errors=True) result['cleanup'] = 'broker 进程已终止、临时目录已删除（Kafka 数据目录一并删除）；未修改本机既有服务。' EVIDENCE.parent.mkdir(parents=True, exist_ok=True) tmp = EVIDENCE.with_suffix('.tmp') tmp.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n') tmp.replace(EVIDENCE) else: result['cleanup'] = '失败运行保留临时目录供排查: ' + str(tmp_root) print(json.dumps(result, ensure_ascii=False, indent=2)) print(json.dumps(result, ensure_ascii=False, indent=2)) return exit_code if result['status'] != 'passed' else 0 if __name__ == '__main__': sys.exit(main())