#!/usr/bin/env python3 """Fetch fixed-tag official sources; cache outside the repo and verify symbols. python3 labs/middleware-senior/verify_sources.py python3 labs/middleware-senior/verify_sources.py --show 'lastStableOffset|tryComplete' Network required. A hit proves existence, not the entire semantic claim. """ import argparse import concurrent.futures import hashlib from pathlib import Path import re import tempfile import urllib.request SOURCES = { "apache/kafka/3.7.1": { "clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java": ["doSend", "RecordAccumulator"], "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java": ["completeBatch", "canRetry"], "clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java": ["sequenceNumber", "ProducerIdAndEpoch"], "core/src/main/scala/kafka/server/KafkaApis.scala": ["handleProduceRequest"], "core/src/main/scala/kafka/server/ReplicaManager.scala": ["appendRecords", "appendToLocalLog"], "core/src/main/scala/kafka/cluster/Partition.scala": ["appendRecordsToLeader", "maybeIncrementLeaderHW", "checkEnoughReplicasReachOffset"], "core/src/main/scala/kafka/server/DelayedProduce.scala": ["tryComplete", "requiredOffset"], "core/src/main/scala/kafka/log/UnifiedLog.scala": ["appendAsLeader", "lastStableOffset", "firstUnstableOffset"], "storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerAppendInfo.java": ["checkSequence", "OutOfOrderSequenceException"], "clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java": ["onJoinPrepare", "commitOffsetsSync", "onJoinComplete"], "clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java": ["pollHeartbeat", "pollTimeoutExpired"], "core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala": ["handleCommitOffsets", "ILLEGAL_GENERATION"], "clients/src/main/java/org/apache/kafka/clients/consumer/CooperativeStickyAssignor.java": ["ownedPartitions", "partitionsTransferringOwnership"], "core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala": ["handleEndTransaction", "PrepareCommit", "PrepareAbort"], }, "apache/zookeeper/release-3.9.2": { "zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java": ["pRequest2Txn", "checkAndIncVersion"], "zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ProposalRequestProcessor.java": ["propose", "commit"], "zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java": ["tryToCommit", "ACK"], "zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java": ["processRequest", "getData"], "zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java": ["killSession", "deleteNode", "triggerWatch"], "zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java": ["EventThread", "Expired"], }, "alibaba/nacos/2.4.3": { "client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxy.java": ["registerService", "redoService"], "naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/InstanceRequestHandler.java": ["registerInstance"], "naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/EphemeralClientOperationServiceImpl.java": ["registerInstance", "ClientRegisterServiceEvent"], "naming/src/main/java/com/alibaba/nacos/naming/consistency/ephemeral/distro/v2/DistroClientDataProcessor.java": ["Distro", "isEphemeral"], "core/src/main/java/com/alibaba/nacos/core/distributed/raft/JRaftProtocol.java": ["CPProtocol"], "client/src/main/java/com/alibaba/nacos/client/config/impl/ClientWorker.java": ["checkListenCache", "executeConfigListen"], "config/src/main/java/com/alibaba/nacos/config/server/service/ConfigOperationService.java": ["publishConfig", "ConfigDataChangeEvent"], }, "apache/dubbo/dubbo-3.2.15": { "dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java": ["invoke", "RpcInvocation"], "dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvoker.java": ["doInvoke", "isBiz", "RETRIES_KEY"], "dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java": ["select", "loadbalance"], "dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java": ["TimeoutCheckTask", "cancel", "CHANNEL_INACTIVE"], "dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java": ["doInvoke", "timeout"], "dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java": ["notify", "refreshOverrideAndInvoker"], }, "openjdk/jdk17u/jdk-17.0.12+7": { "src/java.base/share/classes/sun/nio/ch/SocketChannelImpl.java": ["read", "write"], "src/java.base/linux/classes/sun/nio/ch/EPollSelectorImpl.java": ["doSelect", "processEvents"], "src/java.base/share/classes/java/net/InetAddress.java": ["getAllByName", "cache"], "src/java.base/share/classes/java/util/concurrent/ScheduledThreadPoolExecutor.java": ["decorateTask", "reExecutePeriodic"], }, } def fetch(item): repo, path, symbols = item url = f"https://raw.githubusercontent.com/{repo}/{path}" cache = Path(tempfile.gettempdir()) / "middleware-senior-source-cache" cache.mkdir(exist_ok=True) dest = cache / (hashlib.sha256(url.encode()).hexdigest() + ".txt") if not dest.exists(): with urllib.request.urlopen(url, timeout=45) as response: dest.write_bytes(response.read()) source = dest.read_text() missing = [symbol for symbol in symbols if symbol not in source] return repo, path, source, missing def main(): parser = argparse.ArgumentParser() parser.add_argument("--show", help="Print source context matching this regex") parser.add_argument("--only", default="", help="Filter repository/file paths by regex") args = parser.parse_args() items = [(repo, path, symbols) for repo, files in SOURCES.items() for path, symbols in files.items() if re.search(args.only, repo + "/" + path)] if not items: parser.error('No source files matched') failed = 0 with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: futures = [pool.submit(fetch, item) for item in items] for future in futures: try: repo, path, source, missing = future.result() print(("FAIL" if missing else "OK"), repo, path, "missing=" + str(missing) if missing else "") failed += bool(missing) if args.show: lines = source.splitlines() indices = set() for i, line in enumerate(lines): if re.search(args.show, line): indices.update(range(max(0, i - 3), min(len(lines), i + 9))) for i in sorted(indices): print(f" {i + 1}: {lines[i]}") except Exception as exc: print("FETCH ERROR", str(exc)) failed += 1 raise SystemExit(1 if failed else 0) if __name__ == "__main__": main()