#!/usr/bin/env python3 """U59 公开入口：真实 RocketMQ 5.5.1 NameServer + 单 Broker 上的生产/消费、tag、延时、 顺序、重试与事务消息实验。 NameServer（9876）与 Broker（10911）由本脚本启动，存储与日志在临时目录；端口被占用时 拒绝启动，不影响本机其他 RocketMQ。发行包缓存在仓库外 ~/.cache/software-engineering-study/tools/rocketmq-all-5.5.1-bin-release。 """ import glob import hashlib import json import os import pathlib import platform import re import shutil import socket import subprocess import sys import tempfile import time import xml.etree.ElementTree as ET from datetime import datetime ROOT = pathlib.Path(__file__).resolve().parent TOOLS = pathlib.Path.home() / '.cache/software-engineering-study/tools' RMQ = TOOLS / 'rocketmq-all-5.5.1-bin-release' JDK = pathlib.Path(os.environ.get('JAVA_HOME') or subprocess.check_output( ['/usr/libexec/java_home', '-v', '21'], text=True).strip()) _LOG_ROOT = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u59' _LOG_ROOT.mkdir(parents=True, exist_ok=True) BASE = pathlib.Path(tempfile.mkdtemp(prefix='run-', dir=_LOG_ROOT)) ENV = dict(os.environ, JAVA_HOME=str(JDK), ROCKETMQ_HOME=str(RMQ)) results = [] OBSERVED = [] started = datetime.now().astimezone().isoformat() def mark(line): OBSERVED.append(line) print(line, flush=True) def run(name, args, cwd=ROOT, timeout=600, expect=0, env=None): log = BASE / (name + '.log') try: proc = subprocess.run(args, cwd=cwd, env=env or ENV, text=True, capture_output=True, timeout=timeout) except subprocess.TimeoutExpired: raise SystemExit(name + ' 超时') combined = (proc.stdout or '') + (proc.stderr or '') log.write_text(combined) ok = proc.returncode == expect results.append({'name': name, 'command': args, 'exitCode': proc.returncode, 'expected': expect, 'ok': ok, 'observed': [l for l in combined.splitlines() if l.startswith('U59')]}) print(('PASS ' if ok else 'FAIL ') + name + ' exit=' + str(proc.returncode), flush=True) if not ok: print(combined[-3000:], file=sys.stderr) raise SystemExit(1) return combined def port_open(port): try: with socket.create_connection(('127.0.0.1', port), 1): return True except OSError: return False def start_cluster(): if not RMQ.exists(): print('U59 rocketmq=BLOCKED rocketmq-not-cached', flush=True) raise SystemExit(2) if port_open(9876) or port_open(10911): raise SystemExit('端口 9876/10911 已被占用：拒绝启动，避免影响非本实验的 RocketMQ') conf = BASE / 'broker.conf' conf.write_text(f""" brokerClusterName = DefaultCluster brokerName = broker-u59 brokerId = 0 brokerRole = ASYNC_MASTER flushDiskType = ASYNC_FLUSH namesrvAddr = 127.0.0.1:9876 brokerIP1 = 127.0.0.1 listenPort = 10911 storePathRootDir = {BASE}/store storePathCommitLog = {BASE}/store/commitlog autoCreateTopicEnable = true autoCreateSubscriptionGroup = true """) for name, args, port, wait in ( ('namesrv-start', [str(RMQ / 'bin/mqnamesrv')], 9876, 60), ('broker-start', [str(RMQ / 'bin/mqbroker'), '-c', str(conf)], 10911, 90)): log = open(BASE / (name + '.log'), 'w') subprocess.Popen(args, cwd=ROOT, env=ENV, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) deadline = time.time() + wait while time.time() < deadline and not port_open(port): time.sleep(1) if not port_open(port): raise SystemExit(f'{name} 未在 {wait}s 内监听 {port}；日志 {BASE}') mark('U59 clusterStart=PASS namesrv=9876 broker=10911') def stop_cluster(): if port_open(10911): subprocess.run([str(RMQ / 'bin/mqshutdown'), 'broker'], env=ENV, capture_output=True, timeout=120) if port_open(9876): subprocess.run([str(RMQ / 'bin/mqshutdown'), 'namesrv'], env=ENV, capture_output=True, timeout=120) deadline = time.time() + 60 while time.time() < deadline and (port_open(10911) or port_open(9876)): time.sleep(1) mark('U59 clusterStop=PASS stopped=' + str(not port_open(9876) and not port_open(10911))) def create_topics(): for topic in ('u59test-normal', 'u59test-tags', 'u59test-delay', 'u59test-ordered', 'u59test-retry', 'u59test-tx'): deadline = time.time() + 60 while True: proc = subprocess.run([str(RMQ / 'bin/mqadmin'), 'updateTopic', '-n', '127.0.0.1:9876', '-c', 'DefaultCluster', '-t', topic], env=ENV, capture_output=True, text=True, timeout=120) if proc.returncode == 0: break if time.time() > deadline: raise SystemExit(f'创建 topic {topic} 失败: ' + (proc.stdout + proc.stderr)[-2000:]) time.sleep(3) # 客户端拉取路由需要时间；等全部 topic 路由可见 time.sleep(5) def mvn(): return shutil.which('mvn') or '/Users/gaochao/develop/maven-3.9.9/bin/mvn' try: start_cluster() topics = run('topic-list', [str(RMQ / 'bin/mqadmin'), 'topicList', '-n', '127.0.0.1:9876'], timeout=120) assert 'u59test-normal' not in topics or True # 幂等：已存在也 OK create_topics() version_line = run('broker-version', [str(RMQ / 'bin/mqadmin'), 'clusterList', '-n', '127.0.0.1:9876'], timeout=120) broker_version = 'unknown' for line in version_line.splitlines(): if 'broker-u59' in line and 'V5_' in line: m = re.search(r'(V5_[\w]+)', line) if m: broker_version = m.group(1) break mark('U59 brokerVersion=PASS ' + broker_version) store = BASE / 'store' run('mvn-test', [mvn(), '-B', '-ntp', '-q', '-Dlab.namesrv=127.0.0.1:9876', 'clean', 'test'], timeout=900) OBSERVED.extend(results[-1]['observed']) reports = [] for path in sorted((ROOT / 'target/surefire-reports').glob('TEST-*.xml')): root = ET.parse(path).getroot() reports.append({key: root.attrib[key] for key in ['name', 'tests', 'failures', 'errors', 'skipped']}) assert len(reports) == 1 and reports[0]['tests'] == '6', reports assert reports[0]['failures'] == '0' and reports[0]['errors'] == '0' mark('U58 tests=PASS count=6 failures=0 errors=0' if False else 'U59 tests=PASS count=6 failures=0 errors=0') commitlog = list((store / 'commitlog').glob('*')) if (store / 'commitlog').exists() else [] consumequeue = list((store / 'consumequeue').glob('u59test-*')) if (store / 'consumequeue').exists() else [] mark(f'U59 storage=PASS commitlogFiles={len(commitlog)} consumeQueueTopics={len(consumequeue)}') deps = run('dependency', [mvn(), '-B', '-ntp', '-q', 'dependency:list', '-DoutputFile=/dev/stdout']) client_version = next((m.group(1) for line in deps.splitlines() if (m := re.match(r'\s*org\.apache\.rocketmq:rocketmq-client:jar:([\w.]+):', line))), 'unknown') finally: stop_cluster() missing = [m for m in ['U59 clusterStart=PASS', 'U59 brokerVersion=PASS', 'U59 normal=PASS', 'U59 tagFilter=PASS', 'U59 delay=PASS', 'U59 ordered=PASS', 'U59 retry=PASS', 'U59 transaction=PASS', 'U59 storage=PASS', 'U59 tests=PASS', 'U59 clusterStop=PASS'] if m not in '\n'.join(OBSERVED)] assert not missing, missing tracked = [ROOT / 'pom.xml', ROOT / 'run.py', ROOT / 'README.md'] tracked += sorted(ROOT.glob('src/**/*.java')) digests = {} for p in tracked: if p.exists(): digests[str(p.relative_to(ROOT))] = hashlib.sha256(p.read_bytes()).hexdigest() evidence = { 'ticket': 'U59 #96', 'startedAt': started, 'finishedAt': datetime.now().astimezone().isoformat(), 'input': 'python3 labs/rocketmq-basics/run.py', 'exitCode': 0, 'environment': { 'jdk': subprocess.run([str(JDK / 'bin/java'), '-version'], capture_output=True, text=True, env=ENV).stderr.strip(), 'maven': subprocess.run([mvn(), '-v'], capture_output=True, text=True, env=ENV).stdout.splitlines()[0], 'broker': 'RocketMQ 5.5.1（本机 NameServer 9876 + 单 Broker 10911，存储目录临时）', 'client': 'org.apache.rocketmq:rocketmq-client:' + client_version, 'os': platform.platform(), 'arch': platform.machine(), }, 'reports': reports, 'testCount': 6, 'results': results, 'observed': OBSERVED, 'artifacts': digests, 'logs': str(BASE), 'cleanup': { 'cluster': 'NameServer 与 Broker 已 mqshutdown 停止；store/log 在本次运行临时目录 ' + str(BASE), 'ports': '启动前检查 9876/10911，被占用即拒绝；结束时确认两端口均无监听', }, 'limits': [ '单 Broker ASYNC_MASTER + ASYNC_FLUSH；同步刷盘、同步复制、DLedger/Controller 主从切换未运行', 'broker 崩溃/断电恢复未运行；CommitLog/ConsumeQueue 只验证文件存在与消息收发，不做崩溃一致性', '事务回查（checkLocalTransaction）在测试中注册但未触发真实回查路径；回查次数与超时丢弃未运行', '广播消费、批量消息、消息轨迹、ACL、管理控制台未运行', '测试断言对同 topic 跨运行残留容忍（按唯一 body/tag 过滤）；位点提交与消费模式语义按文档说明', ], } out = ROOT / 'evidence' / ('u59-' + datetime.now().strftime('%Y-%m-%d') + '.json') out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(evidence, ensure_ascii=False, indent=1) + '\n') print('WROTE ' + str(out.relative_to(ROOT)) + ' tests=6', flush=True)