Apache Cassandra 分布式测试ClusterUtils 完整参考与实战指南【免费下载链接】cassandraOpen source transactional distributed database. Linear scalability and proven fault-tolerance on commodity hardware or cloud infrastructure without compromising performance.项目地址: https://gitcode.com/GitHub_Trending/cassa/cassandra导读ClusterUtils是 Apache Cassandra in-JVM 分布式测试jvm-dtest框架中最重要的工具类提供节点生命周期管理、集群元数据CMS控制、Ring/Gossip 监控、目录管理、节点替换/退役等一整套测试辅助能力。本指南以该类的完整参考文档为主体结合 ClusterUtils.java 的源码实现展开帮助你在不启动真实网络的情况下用单 JVM 多类加载器模拟多节点 Cassandra 集群并精确编排拓扑变更、故障注入与 Accord/CMS 时序场景。读完本文你将能熟练使用ClusterUtils编写健壮、可复现的分布式集成测试。1. 类定位与使用前提ClusterUtils位于org.apache.cassandra.distributed.shared包源码见 ClusterUtils.java。该类被标记为Isolated原因在于它依赖的 lambda 位于被标记为 shared 的包中需要告知 jvm-dtest 不要共享该类同时类注释明确要求该类永远只能在 App ClassLoader测试侧中调用绝不能从集群内部调用。这意味着所有需要触碰节点内部状态的逻辑都通过IInvokableInstance.callOnInstance(...)/runOnInstance(...)以可序列化 lambda 的形式跨类加载器执行。它配合 SKILL.md 中介绍的Cluster、AbstractCluster、InstanceConfig、MessageFilters等构件使用是编写拓扑变更类测试的标配工具。2. 生命周期管理Lifecycle Management生命周期管理是 jvm-dtest 中最频繁使用的 API控制节点的启动、停止与重启。2.1 start()携带系统属性启动I extends IInstance I start(I inst, ConsumerWithProperties fn) I extends IInstance I start(I inst, BiConsumerI, WithProperties fn)启动实例并附带一组系统属性实例启动完成后这些属性会被自动清除。源码实现ClusterUtils.java利用WithProperties的 try-with-resources 语义保证属性生命周期可控ClusterUtils.start(instance, properties - { properties.set(RING_DELAY, 5000); properties.set(BROADCAST_INTERVAL_MS, 30000); });BiConsumerI, WithProperties变体还允许在启动前直接操作实例对象本身适合需要先改配置再启动的场景。2.2 stopUnchecked()阻塞式优雅停机void stopUnchecked(IInstance i)阻塞式停止实例异常以运行时异常抛出。源码为Futures.getUnchecked(i.shutdown())L156-L159与直接调用IInstance#shutdown()的主要区别在于它会捕获 Future 等待过程中的异常并转抛为 RuntimeException避免测试代码到处处理 checked exception。2.3 stopAbrupt()模拟 kill -9I extends IInstance void stopAbrupt(IClusterI cluster, I inst)模拟进程被kill -9强杀实现上先通过cluster.filters().allVerbs().to(inst.config().num()).drop()与.from(...)屏蔽该节点所有进出消息使集群中其他节点无法与其通信然后再执行优雅停机L171-L185。注意如果该实例随后被重启集群会发现它是正常关闭的与真实 kill -9 不完全等价。2.4 stopAll()停止全部实例I extends IInstance void stopAll(IClusterI cluster)停止集群中所有实例但不清理集群状态——这与ICluster#close()不同后者会释放集群资源。2.5 restartUnchecked()阻塞式重启void restartUnchecked(IInstance instance)先stopUnchecked再startup()全程阻塞L201-L208。适合重启后验证状态恢复的测试。3. 实例管理Instance Management3.1 addInstance()创建新实例不启动I extends IInstance I addInstance(AbstractClusterI cluster, ConsumerIInstanceConfig fn) I extends IInstance I addInstance(AbstractClusterI cluster) I extends IInstance I addInstance(AbstractClusterI cluster, IInstanceConfig other, ConsumerIInstanceConfig fn) I extends IInstance I addInstance(AbstractClusterI cluster, String dc, String rack) I extends IInstance I addInstance(AbstractClusterI cluster, String dc, String rack, ConsumerIInstanceConfig fn)创建一个新实例并加入集群但不启动。源码显示L218-L294不带other参数时从集群中第一个未关闭实例复制配置指定 dc/rack 时会把新实例地址写入config.networkTopology()源码 TODO 注释提示当前要求创建集群时已知所有实例否则 NetworkTopology/TokenStrategy 无法识别该行为较隐晦最终调用cluster.bootstrap(config)返回新实例。// 与现有节点相同配置 IInstance inst ClusterUtils.addInstance(cluster); // 自定义配置 IInstance inst ClusterUtils.addInstance(cluster, config - { config.set(auto_bootstrap, true); config.set(num_tokens, 256); }); // 指定 DC/rack IInstance inst ClusterUtils.addInstance(cluster, dc1, rack1, config - { config.set(concurrent_reads, 64); });结合 SKILL.md 中的 NodeAdditionTest 模式addInstance之后调用newNode.startup()再用awaitRingJoin等待入环。3.2 replaceHostAndStart()替换节点并启动I extends IInstance I replaceHostAndStart(AbstractClusterI cluster, I toReplace) I extends IInstance I replaceHostAndStart(AbstractClusterI cluster, I toReplace, ConsumerWithProperties fn) I extends IInstance I replaceHostAndStart(AbstractClusterI cluster, I toReplace, BiConsumerI, WithProperties fn) I extends IInstance I replaceHostAndStart(AbstractClusterI cluster, I toReplace, BiConsumerI, WithProperties fn, ConsumerIInstanceConfig configFn)创建并启动一个新实例来替换既有实例新实例与被替换实例位于同一 DC/rack。源码L347-L359中自动设置auto_bootstraptrueprogress_barrier_min_consistency_levelONE降低引导期间一致性屏障要求以加速测试之后执行startHostReplacement。IInvokableInstance replacement ClusterUtils.replaceHostAndStart( cluster, failedNode, (inst, properties) - { properties.set(RING_DELAY, 5000); properties.set(BROADCAST_INTERVAL_MS, 30000); }, config - { config.set(concurrent_compactors, 8); } );3.3 startHostReplacement()以替换模式启动I extends IInstance I startHostReplacement(I toReplace, I inst) I extends IInstance I startHostReplacement(I toReplace, I inst, BiConsumerI, WithProperties fn)这是replaceHostAndStart的底层启动逻辑L375-L391会设置一组替换专用属性BROADCAST_INTERVAL_MS30s降低广播间隔减少等待RING_DELAY10s默认 30s测试中降低以提速BOOTSTRAP_SCHEMA_DELAY_MS10sREPLACE_ADDRESS_FIRST_BOOTtoReplace的broadcastAddress:port指明被替换节点地址jvm-dtest 中端口可能变化必须显式带上端口。4. Token 与元数据Token Metadata4.1 获取 Token方法签名语义getTokenMetadataTokensListString getTokenMetadataTokens(IInvokableInstance inst)从ClusterMetadata.current().tokenMap.tokens()获取全部 token 字符串列表getLocalTokensCollectionString getLocalTokens(IInvokableInstance inst)通过tokenMap.tokens(myNodeId)获取本节点持有的 tokengetTokensListString getTokens(IInstance instance)从配置文件读取num_tokens与initial_token仅对显式配置了 token的实例有效对学习/自动生成的 token 不适用getTokenCountint getTokenCount(IInvokableInstance instance)读取配置num_tokensgetPartitionerNameString getPartitionerName(IInstance instance)读取配置partitioner源码细节getTokensL1328-L1336断言num_tokens必须为 1 且initial_token非空否则直接失败——这是它只支持显式配置限制的具体体现。4.2 getPrimaryRanges()主副本范围ListRange getPrimaryRanges(IInvokableInstance instance, String keyspace)返回指定 keyspace 下该实例的主 token 范围底层调用TokenRingUtils.getPrimaryRangesForEndpoint(keyspace, broadcastAddress)并对范围做 unwrap 后包装成可序列化的ClusterUtils.RangeL1793-L1801ListClusterUtils.Range ranges ClusterUtils.getPrimaryRanges( cluster.get(1), my_keyspace ); for (ClusterUtils.Range range : ranges) { System.out.println(range.left() to range.right()); }5. 集群元数据与 CMSCluster Metadata CMSCassandra 的 TCMTransformation-based Cluster Metadata通过 CMSCluster Metadata Service节点协调元数据变更每个变更对应一个单调递增的Epoch。ClusterUtils提供整套 epoch 观测与同步工具。5.1 Epoch 观测Epoch getClusterMetadataVersion(IInvokableInstance inst) // 当前集群元数据 epoch Epoch getCurrentEpoch(IInvokableInstance inst) // ClusterMetadata.current().epoch Epoch getNextEpoch(IInvokableInstance inst) // ClusterMetadata.current().nextEpoch() Epoch maxEpoch(IClusterIInvokableInstance cluster) // 全集群最大 epoch跳过关闭实例 Epoch maxEpoch(IClusterIInvokableInstance cluster, int... nodes) // 指定节点中的最大 epochmaxEpochL711-L739在无节点可查时会抛出AssertionError可作为集群已无存活节点的隐式断言。5.2 waitForCMSToQuiesce()等待 CMS 收敛void waitForCMSToQuiesce(IClusterIInvokableInstance cluster, int... cmsNodes) void waitForCMSToQuiesce(IClusterIInvokableInstance cluster, Epoch awaitedEpoch, int... ignored) void waitForCMSToQuiesce(IClusterIInvokableInstance cluster, Epoch awaitedEpoch, boolean fetchLogWhenBehind, int... ignored)等待所有节点ignored指定的除外的集群元数据达到同一状态。实现L741-L778带 30 秒 deadline 轮询间隔 10ms收集落后节点若fetchLogWhenBehindtrue且节点 epoch 落后则先调用fetchLogFromCMS主动拉取日志再比较。超时抛出带明细的AssertionError// 等待所有节点收敛 ClusterUtils.waitForCMSToQuiesce(cluster); // 等待指定 epoch Epoch targetEpoch ClusterUtils.maxEpoch(cluster); ClusterUtils.waitForCMSToQuiesce(cluster, targetEpoch);实际测试中AlterTopologyTest.java 即在拓扑变更后调用waitForCMSToQuiesce(cluster, cmsInstance)等待元数据应用完成。5.3 fetchLogFromCMS() / snapshotClusterMetadata()Epoch fetchLogFromCMS(IInvokableInstance inst, Epoch awaitedEpoch) Epoch fetchLogFromCMS(IInvokableInstance inst, long awaitedEpoch) Epoch snapshotClusterMetadata(IInvokableInstance inst)fetchLogFromCMS通过ClusterMetadataService.instance().fetchLogFromCMS(Epoch)从 CMS 拉取日志直到目标 epochL780-L789snapshotClusterMetadata触发triggerSnapshot()并返回快照 epochL801-L807。5.4 分布式观测MapString, Epoch getPeerEpochs(IInvokableInstance requester) // 向所有 peer 发送 TCM_CURRENT_EPOCH_REQ 汇总各节点 epoch SetString getCMSMembers(IInvokableInstance inst) // fullCMSMembers() 的地址集合getPeerEpochsL809-L829是少见的跨节点主动请求型工具向目录中所有地址发送Verb.TCM_CURRENT_EPOCH_REQ消息并通过 latch 等待全部回包返回peer - Epoch映射。5.5 调试字符串ListString getPeerDirectoryDebugStrings(IInvokableInstance inst) // directory.toDebugString() 按行拆分 ListString getTokenMapDebugStrings(IInvokableInstance inst) // tokenMap.toDebugString() 按行拆分 void logTokenMapDebugString(IInvokableInstance inst) // tokenMap.logDebugString() MapString, List[] getDataPlacementDebugInfo(IInvokableInstance inst) void logDataPlacementDebugString(IInvokableInstance inst, boolean byEndpoint)getDataPlacementDebugInfoL433-L456按 keyspace 返回二元数组下标 0 为读副本列表、下标 1 为写副本列表ReplicaGroups.toReplicaStringList()适合校验数据放置策略。6. CMS 暂停与控制CMS Pausing Control这是ClusterUtils最有价值的故障注入能力通过TestProcessorCMS 提交侧与TestChangeListener节点应用侧在精确时刻定格集群元数据流转构造竞态窗口。6.1 pauseBeforeCommit()提交前暂停CallableEpoch pauseBeforeCommit(IInvokableInstance cmsInstance, SerializablePredicateTransformation predicate)在 CMS 上注册谓词当待提交的Transformation匹配时暂停提交。返回的CallableEpoch会阻塞等待暂停发生并返回暂停时的 epoch。实现L602-L622通过((SwitchableProcessor) ClusterMetadataService.instance().processor()).delegate()取到TestProcessor并调用pauseIf(predicate, ...)超时 30 秒。CallableEpoch pauseHandle ClusterUtils.pauseBeforeCommit( cmsNode, transformation - transformation instanceof AddNode ); // 阻塞直到谓词匹配 Epoch pausedAt pauseHandle.call(); // 暂停期间执行其他操作 // ... // 恢复提交 ClusterUtils.unpauseCommits(cmsNode);6.2 getSequenceAfterCommit()提交后取 epochCallableEpoch getSequenceAfterCommit(IInvokableInstance cmsInstance, SerializableBiPredicateTransformation, Commit.Result predicate)在 CMS 提交侧注册提交谓词当某个 transformation 的提交结果匹配时返回其成功提交后的最新 epochresult.success().logState.latestEpoch()同样 30 秒超时L624-L653。6.3 应用侧暂停pauseBeforeEnacting / pauseAfterEnactingCallableVoid pauseBeforeEnacting(IInvokableInstance instance, long epoch) CallableVoid pauseBeforeEnacting(IInvokableInstance instance, Epoch epoch) CallableVoid pauseAfterEnacting(IInvokableInstance instance, Epoch epoch)在普通节点上暂停应用enact某个 epoch 的前/后时刻L529-L600。实现依赖TestChangeListener.instance通过AsyncPromise通知等待方pauseBeforeEnacting默认超时 30 秒pauseAfterEnacting默认 10 秒CallableVoid pauseHandle ClusterUtils.pauseBeforeEnacting(instance, targetEpoch); pauseHandle.call(); // 等待暂停 // 暂停期间执行操作 // ... ClusterUtils.unpauseEnactment(instance); // 恢复6.4 恢复与清理void unpauseCommits(IInvokableInstance instance) // 恢复 CMS 提交若实例已关闭则静默跳过 void unpauseEnactment(IInvokableInstance instance) // 恢复 epoch 应用 void clearAndUnpause(IInvokableInstance instance) // 清除全部暂停条件并恢复unpauseCommitsL655-L663对已关闭实例做了防护性判断避免在节点宕机后误操作。6.5 日志条目过滤void dropAllEntriesBeginningAt(IInvokableInstance instance, Epoch epoch) void clearEntryFilters(IInvokableInstance instance)dropAllEntriesBeginningAt为 CMS 日志添加过滤器丢弃epoch 指定值的所有条目L519-L522用于模拟元数据日志丢失clearEntryFilters清除全部过滤器。7. Ring 与 Gossip 监控Ring Gossip Monitoring7.1 ring()解析 nodetool ring 输出ListRingInstanceDetails ring(IInstance inst)内部执行inst.nodetoolResult(ring)并断言成功再用正则解析输出L925-L930。RingInstanceDetails字段address节点地址rack机架名statusUp / DownstateNormal / Leaving / Joining / Movingtokentoken 值7.2 Ring 断言立即检查ListRingInstanceDetails assertInRing(IInstance instance, IInstance expectedInRing) ListRingInstanceDetails assertNotInRing(IInstance instance, IInstance expectedInRing) ListRingInstanceDetails assertRingState(IInstance instance, IInstance expectedInRing, String state) ListRingInstanceDetails assertRingIs(IInstance instance, IInstance... expectedInRing) ListRingInstanceDetails assertRingIs(IInstance instance, Collection? extends IInstance expectedInRing) ListRingInstanceDetails assertRingIs(IInstance instance, SetString expectedRingAddresses)assertRingIs校验 ring 中恰好包含期望实例集合源码注释特别说明查询源节点自身可能不在 ring 中因此仅依赖期望集合比对地址使用 AssertJ 的isEqualTo保证严格相等L1134-L1142。7.3 Ring 等待轮询ListRingInstanceDetails awaitRingJoin(IInstance instance, IInstance expectedInRing) ListRingInstanceDetails awaitRingJoin(IInstance instance, String expectedInRing) void awaitRingJoin(Cluster cluster, int[] nodes, IInvokableInstance expectedInRing) ListRingInstanceDetails awaitRingHealthy(IInstance src) ListRingInstanceDetails awaitRingStatus(IInstance instance, IInstance expectedInRing, String status) ListRingInstanceDetails awaitRingState(IInstance instance, IInstance expectedInRing, String state)底层awaitRingL985-L998最多轮询 100 次、每次间隔 1 秒超时抛出带当前 ring 快照的AssertionError。awaitRingJoin判定条件为statusUp stateNormalawaitRingHealthy要求 ring 中所有实例均 Up 且 Normal。7.4 Gossip 信息MapString, MapString, String gossipInfo(IInstance inst) void assertGossipInfo(IInstance instance, InetSocketAddress expectedInGossip, int expectedGeneration, int expectedHeartbeat) MapString, MapString, String awaitGossipStatus(IInstance instance, IInstance expectedInGossip, String targetStatus) void awaitGossipSchemaMatch(ICluster? extends IInstance cluster) void awaitGossipSchemaMatch(IInstance instance) void awaitGossipStateMatch(ICluster? extends IInstance cluster, IInstance expectedInGossip, ApplicationState key)gossipInfo执行nodetool gossipinfo并解析为地址 - (generation, heartbeat, STATUS, ...)映射L1271-L1319。awaitGossipStatus兼容STATUS_WITH_PORT与旧版STATUS两个键使用 contains 匹配如传NORMAL。awaitGossipSchemaMatch会跳过未入环节点比较已入环节点的SCHEMA值是否一致若集群未启用Feature.GOSSIP则直接返回L1213-L1247。8. 系统表System Tablesvoid awaitInPeers(Cluster cluster, int[] nodes, IInstance expectedInPeers) void awaitInPeers(IInstance instance, IInstance expectedInPeers) boolean isInPeers(IInstance instance, IInstance expectedInPeers)isInPeersL1679-L1689通过内部查询SELECT tokens, data_center, rack FROM system.peers WHERE peer?判断目标节点是否已以完整信息出现在 peers 表中tokens 非空、dc/rack 非空。awaitInPeers轮询该条件最多 100 秒集群级重载会跳过自身节点节点不会出现在自己的 peers 中。9. 目录与文件Directories Files9.1 目录获取ListFile getDataDirectories(IInstance instance) File getCommitLogDirectory(IInstance instance) File getHintsDirectory(IInstance instance) File getSavedCachesDirectory(IInstance instance) File getJournalDirectory(IInstance instance) // Accord journal 目录配置项 accord.journal_directory File getCdcRawDirectory(IInstance instance) ListFile getDirectories(IInstance instance) // data commitlog hints saved caches journal cdc实现直接从IInstanceConfig读取对应配置项如data_file_directories、commitlog_directory、accord.journal_directory、cdc_raw_directory等源码注释坦承这依赖InstanceConfig的具体实现L1355-L1442。9.2 cleanup()重置实例到干净状态void cleanup(IInvokableInstance inst)依次执行stopUnchecked→ 删除全部可写目录递归→startup()L1449-L1458。等价于完全重来适用于需要清空数据重跑的测试。10. 节点操作Node Operations10.1 decommission()boolean decommission(IInvokableInstance leaving)在目标节点内部调用StorageService.instance.decommission(true)返回是否成功L856-L870boolean success ClusterUtils.decommission(cluster.get(3)); Assert.assertTrue(Decommission failed, success); ClusterUtils.assertNotInRing(cluster.get(1), cluster.get(3));10.2 getNodeId() / cancelInProgressSequences()NodeId getNodeId(IInvokableInstance target) int getNodeId(IInvokableInstance target, IInvokableInstance executor) boolean cancelInProgressSequences(IInvokableInstance executor) boolean cancelInProgressSequences(NodeId nodeId, IInvokableInstance executor)getNodeId通过executor节点查询ClusterMetadata.current().directory.peerId(target地址)得到 TCM 视角的NodeIdL872-L891。cancelInProgressSequences调用StorageService.instance.cancelInProgressSequences(...)用于清理节点加入/离开过程中残留的进行中序列。10.3 mode() / assertModeJoined() / isMigrating()StorageService.Mode mode(IInvokableInstance inst) void assertModeJoined(IInvokableInstance inst) boolean isMigrating(IInvokableInstance instance)mode返回StorageService.Mode可取STARTING、NORMAL、JOINING、LEAVING、DECOMMISSIONED、MOVING、DRAINING、DRAINEDL1691-L1695assertModeJoined断言模式为NORMALisMigrating查询ClusterMetadataService.instance().isMigrating()判断 CMS 是否正在迁移L675-L678。11. 日志监控Log Monitoringvoid runAndWaitForLogs(Runnable r, String waitString, AbstractClusterI cluster) void runAndWaitForLogs(Runnable r, String waitString, IInstance... instances)先在每个目标实例上logs().mark()标记位置执行操作再逐个watchFor(mark, waitString)等待日志出现L484-L497ClusterUtils.runAndWaitForLogs( () - cluster.get(2).nodetool(bootstrap), Bootstrap completed, cluster.get(1), cluster.get(2), cluster.get(3) );12. 地址管理Address Managementvoid updateAddress(IInstance instance, String address) String getBroadcastAddressHostWithPortString(IInstance target) InetSocketAddress getNativeInetSocketAddress(IInstance target) int getIntConfig(IInstanceConfig config, String configName, int defaultValue)updateAddressL1487-L1530同时修改broadcast_address、listen_address、broadcast_rpc_address、rpc_address四个配置项只能在该实例处于关闭状态时调用否则行为未定义。实现细节包括强制清空InstanceConfig缓存的InetSocketAddress - InetAddressAndPort映射否则启动会忽略新配置并通过反射更新NetworkTopology内部的 HashMap源码 TODO 提示 NetworkTopology 不支持动态增删且非线程安全节点运行期间变更存在风险。getNativeInetSocketAddress返回 CQL 原生协议地址端口读取native_transport_port缺省回退 9042L1552-L1556getIntConfig在配置缺失NPE时返回默认值L1575-L1585。13. Accord 测试辅助13.1 事务状态查询T extends IInstance LinkedHashMapString, SimpleQueryResult queryTxnState( AbstractClusterT cluster, TxnId txnId, int... nodes) T extends IInstance String queryTxnStateAsString(AbstractClusterT cluster, TxnId txnId, int... nodes) T extends IInstance void queryTxnStateAsString(StringBuilder sb, AbstractClusterT cluster, TxnId txnId, int... nodes)queryTxnStateL1704-L1717对每个目标节点执行SELECT * FROM system_accord.TXN_BLOCKED_BY WHERE txn_id?SchemaConstants.VIRTUAL_ACCORD_DEBUGAccordDebugKeyspace.TXN_BLOCKED_BY跳过已关闭节点nodes为空时遍历整个集群。queryTxnStateAsString将结果拼为可读文本AccordTestBase.java 中即用它输出故障现场。13.2 tableId() / awaitAccordEpochReady()TableId tableId(Cluster cluster, String ks, String table) void awaitAccordEpochReady(Cluster cluster, long epoch)tableId通过Schema.instance.getKeyspaceInstance(ks).getColumnFamilyStore(table).getTableId()获取表 IDL1743-L1747awaitAccordEpochReady等待所有存活实例上AccordService.instance().epochReady(Epoch, EpochReady::reads)完成即指定 epoch 的读路径就绪L1749-L1764。14. 工具类Utility Classes14.1 Rangepublic static class Range implements Serializable { public final String left, right; public Range(String left, String right) public Range(long left, long right) public long left() public long right() }表示一个 token 范围标记为Shared可跨类加载器序列化。getPrimaryRanges返回的即为此类实例。14.2 RingInstanceDetailspublic static final class RingInstanceDetails { private final String address; // 节点地址 private final String rack; // 机架名 private final String status; // Up/Down private final String state; // Normal/Leaving/Joining/Moving private final String token; // token 值 public String getAddress() / getRack() / getStatus() / getState() / getToken() }ring()/awaitRing*/assert*系列返回的解析结果对象实现完整的equals/hashCode/toStringL1587-L1652可直接用于断言比较。15. 实战组合一个完整的节点替换测试综合以上 API一个典型的替换故障节点测试流程如下模式见 SKILL.md 及 HostReplacementTest.java 等用例try (Cluster cluster Cluster.build(3).start()) { // 1. 创建 keyspace/table省略 // 2. 模拟节点 2 被 kill -9 IInvokableInstance original cluster.get(2); ClusterUtils.stopAbrupt(cluster, original); // 3. 创建并启动替换节点 IInvokableInstance replacement ClusterUtils.replaceHostAndStart( cluster, original, (inst, properties) - properties.set(RING_DELAY, 5000)); // 4. 等待元数据收敛CMS quiesce ClusterUtils.waitForCMSToQuiesce(cluster); // 5. 等待替换节点入环 ClusterUtils.awaitRingJoin(cluster.get(1), replacement); ClusterUtils.awaitRingHealthy(cluster.get(1)); // 6. 校验 ring 成员关系 ClusterUtils.assertInRing(cluster.get(1), replacement); ClusterUtils.assertNotInRing(cluster.get(1), original); }16. 最佳实践小结只在测试侧调用ClusterUtils是Isolated类禁止在节点内部runOnInstance的 lambda 内调用所有内部逻辑走callOnInstance/runOnInstance。善用等待而非 sleep优先使用awaitRingJoin、awaitRingHealthy、waitForCMSToQuiesce、awaitGossipSchemaMatch等带轮询超时的方法它们会在超时时给出包含现场快照的AssertionError便于定位。区分断言与等待assert*系列立即检查不满足即失败await*系列轮询等待默认约 100 秒、1 秒间隔二者应组合使用。CMS 暂停是竞态利器pauseBeforeCommit/pauseBeforeEnacting配合unpauseCommits/unpauseEnactment可在拓扑变更的精确时机注入竞争使用后务必恢复或调用clearAndUnpause。注意限制getTokens仅适用于显式配置 token 的实例updateAddress仅能在节点关闭时调用addInstance要求集群创建时就规划好拓扑。替换/退役后必须收敛replaceHostAndStart/decommission之后用waitForCMSToQuiesceawaitRingJoin/assertNotInRing完成双维度校验元数据 epoch 与 ring 视图。通过将本指南与 SKILL.md、advanced_patterns.md、classloader_guide.md 配合使用你可以在单 JVM 内高效复现多节点集群的各类拓扑与故障场景。【免费下载链接】cassandraOpen source transactional distributed database. Linear scalability and proven fault-tolerance on commodity hardware or cloud infrastructure without compromising performance.项目地址: https://gitcode.com/GitHub_Trending/cassa/cassandra创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Cassandra 分布式测试：ClusterUtils 完整参考与实战指南
Apache Cassandra 分布式测试ClusterUtils 完整参考与实战指南【免费下载链接】cassandraOpen source transactional distributed database. Linear scalability and proven fault-tolerance on commodity hardware or cloud infrastructure without compromising performance.项目地址: https://gitcode.com/GitHub_Trending/cassa/cassandra导读ClusterUtils是 Apache Cassandra in-JVM 分布式测试jvm-dtest框架中最重要的工具类提供节点生命周期管理、集群元数据CMS控制、Ring/Gossip 监控、目录管理、节点替换/退役等一整套测试辅助能力。本指南以该类的完整参考文档为主体结合 ClusterUtils.java 的源码实现展开帮助你在不启动真实网络的情况下用单 JVM 多类加载器模拟多节点 Cassandra 集群并精确编排拓扑变更、故障注入与 Accord/CMS 时序场景。读完本文你将能熟练使用ClusterUtils编写健壮、可复现的分布式集成测试。1. 类定位与使用前提ClusterUtils位于org.apache.cassandra.distributed.shared包源码见 ClusterUtils.java。该类被标记为Isolated原因在于它依赖的 lambda 位于被标记为 shared 的包中需要告知 jvm-dtest 不要共享该类同时类注释明确要求该类永远只能在 App ClassLoader测试侧中调用绝不能从集群内部调用。这意味着所有需要触碰节点内部状态的逻辑都通过IInvokableInstance.callOnInstance(...)/runOnInstance(...)以可序列化 lambda 的形式跨类加载器执行。它配合 SKILL.md 中介绍的Cluster、AbstractCluster、InstanceConfig、MessageFilters等构件使用是编写拓扑变更类测试的标配工具。2. 生命周期管理Lifecycle Management生命周期管理是 jvm-dtest 中最频繁使用的 API控制节点的启动、停止与重启。2.1 start()携带系统属性启动I extends IInstance I start(I inst, ConsumerWithProperties fn) I extends IInstance I start(I inst, BiConsumerI, WithProperties fn)启动实例并附带一组系统属性实例启动完成后这些属性会被自动清除。源码实现ClusterUtils.java利用WithProperties的 try-with-resources 语义保证属性生命周期可控ClusterUtils.start(instance, properties - { properties.set(RING_DELAY, 5000); properties.set(BROADCAST_INTERVAL_MS, 30000); });BiConsumerI, WithProperties变体还允许在启动前直接操作实例对象本身适合需要先改配置再启动的场景。2.2 stopUnchecked()阻塞式优雅停机void stopUnchecked(IInstance i)阻塞式停止实例异常以运行时异常抛出。源码为Futures.getUnchecked(i.shutdown())L156-L159与直接调用IInstance#shutdown()的主要区别在于它会捕获 Future 等待过程中的异常并转抛为 RuntimeException避免测试代码到处处理 checked exception。2.3 stopAbrupt()模拟 kill -9I extends IInstance void stopAbrupt(IClusterI cluster, I inst)模拟进程被kill -9强杀实现上先通过cluster.filters().allVerbs().to(inst.config().num()).drop()与.from(...)屏蔽该节点所有进出消息使集群中其他节点无法与其通信然后再执行优雅停机L171-L185。注意如果该实例随后被重启集群会发现它是正常关闭的与真实 kill -9 不完全等价。2.4 stopAll()停止全部实例I extends IInstance void stopAll(IClusterI cluster)停止集群中所有实例但不清理集群状态——这与ICluster#close()不同后者会释放集群资源。2.5 restartUnchecked()阻塞式重启void restartUnchecked(IInstance instance)先stopUnchecked再startup()全程阻塞L201-L208。适合重启后验证状态恢复的测试。3. 实例管理Instance Management3.1 addInstance()创建新实例不启动I extends IInstance I addInstance(AbstractClusterI cluster, ConsumerIInstanceConfig fn) I extends IInstance I addInstance(AbstractClusterI cluster) I extends IInstance I addInstance(AbstractClusterI cluster, IInstanceConfig other, ConsumerIInstanceConfig fn) I extends IInstance I addInstance(AbstractClusterI cluster, String dc, String rack) I extends IInstance I addInstance(AbstractClusterI cluster, String dc, String rack, ConsumerIInstanceConfig fn)创建一个新实例并加入集群但不启动。源码显示L218-L294不带other参数时从集群中第一个未关闭实例复制配置指定 dc/rack 时会把新实例地址写入config.networkTopology()源码 TODO 注释提示当前要求创建集群时已知所有实例否则 NetworkTopology/TokenStrategy 无法识别该行为较隐晦最终调用cluster.bootstrap(config)返回新实例。// 与现有节点相同配置 IInstance inst ClusterUtils.addInstance(cluster); // 自定义配置 IInstance inst ClusterUtils.addInstance(cluster, config - { config.set(auto_bootstrap, true); config.set(num_tokens, 256); }); // 指定 DC/rack IInstance inst ClusterUtils.addInstance(cluster, dc1, rack1, config - { config.set(concurrent_reads, 64); });结合 SKILL.md 中的 NodeAdditionTest 模式addInstance之后调用newNode.startup()再用awaitRingJoin等待入环。3.2 replaceHostAndStart()替换节点并启动I extends IInstance I replaceHostAndStart(AbstractClusterI cluster, I toReplace) I extends IInstance I replaceHostAndStart(AbstractClusterI cluster, I toReplace, ConsumerWithProperties fn) I extends IInstance I replaceHostAndStart(AbstractClusterI cluster, I toReplace, BiConsumerI, WithProperties fn) I extends IInstance I replaceHostAndStart(AbstractClusterI cluster, I toReplace, BiConsumerI, WithProperties fn, ConsumerIInstanceConfig configFn)创建并启动一个新实例来替换既有实例新实例与被替换实例位于同一 DC/rack。源码L347-L359中自动设置auto_bootstraptrueprogress_barrier_min_consistency_levelONE降低引导期间一致性屏障要求以加速测试之后执行startHostReplacement。IInvokableInstance replacement ClusterUtils.replaceHostAndStart( cluster, failedNode, (inst, properties) - { properties.set(RING_DELAY, 5000); properties.set(BROADCAST_INTERVAL_MS, 30000); }, config - { config.set(concurrent_compactors, 8); } );3.3 startHostReplacement()以替换模式启动I extends IInstance I startHostReplacement(I toReplace, I inst) I extends IInstance I startHostReplacement(I toReplace, I inst, BiConsumerI, WithProperties fn)这是replaceHostAndStart的底层启动逻辑L375-L391会设置一组替换专用属性BROADCAST_INTERVAL_MS30s降低广播间隔减少等待RING_DELAY10s默认 30s测试中降低以提速BOOTSTRAP_SCHEMA_DELAY_MS10sREPLACE_ADDRESS_FIRST_BOOTtoReplace的broadcastAddress:port指明被替换节点地址jvm-dtest 中端口可能变化必须显式带上端口。4. Token 与元数据Token Metadata4.1 获取 Token方法签名语义getTokenMetadataTokensListString getTokenMetadataTokens(IInvokableInstance inst)从ClusterMetadata.current().tokenMap.tokens()获取全部 token 字符串列表getLocalTokensCollectionString getLocalTokens(IInvokableInstance inst)通过tokenMap.tokens(myNodeId)获取本节点持有的 tokengetTokensListString getTokens(IInstance instance)从配置文件读取num_tokens与initial_token仅对显式配置了 token的实例有效对学习/自动生成的 token 不适用getTokenCountint getTokenCount(IInvokableInstance instance)读取配置num_tokensgetPartitionerNameString getPartitionerName(IInstance instance)读取配置partitioner源码细节getTokensL1328-L1336断言num_tokens必须为 1 且initial_token非空否则直接失败——这是它只支持显式配置限制的具体体现。4.2 getPrimaryRanges()主副本范围ListRange getPrimaryRanges(IInvokableInstance instance, String keyspace)返回指定 keyspace 下该实例的主 token 范围底层调用TokenRingUtils.getPrimaryRangesForEndpoint(keyspace, broadcastAddress)并对范围做 unwrap 后包装成可序列化的ClusterUtils.RangeL1793-L1801ListClusterUtils.Range ranges ClusterUtils.getPrimaryRanges( cluster.get(1), my_keyspace ); for (ClusterUtils.Range range : ranges) { System.out.println(range.left() to range.right()); }5. 集群元数据与 CMSCluster Metadata CMSCassandra 的 TCMTransformation-based Cluster Metadata通过 CMSCluster Metadata Service节点协调元数据变更每个变更对应一个单调递增的Epoch。ClusterUtils提供整套 epoch 观测与同步工具。5.1 Epoch 观测Epoch getClusterMetadataVersion(IInvokableInstance inst) // 当前集群元数据 epoch Epoch getCurrentEpoch(IInvokableInstance inst) // ClusterMetadata.current().epoch Epoch getNextEpoch(IInvokableInstance inst) // ClusterMetadata.current().nextEpoch() Epoch maxEpoch(IClusterIInvokableInstance cluster) // 全集群最大 epoch跳过关闭实例 Epoch maxEpoch(IClusterIInvokableInstance cluster, int... nodes) // 指定节点中的最大 epochmaxEpochL711-L739在无节点可查时会抛出AssertionError可作为集群已无存活节点的隐式断言。5.2 waitForCMSToQuiesce()等待 CMS 收敛void waitForCMSToQuiesce(IClusterIInvokableInstance cluster, int... cmsNodes) void waitForCMSToQuiesce(IClusterIInvokableInstance cluster, Epoch awaitedEpoch, int... ignored) void waitForCMSToQuiesce(IClusterIInvokableInstance cluster, Epoch awaitedEpoch, boolean fetchLogWhenBehind, int... ignored)等待所有节点ignored指定的除外的集群元数据达到同一状态。实现L741-L778带 30 秒 deadline 轮询间隔 10ms收集落后节点若fetchLogWhenBehindtrue且节点 epoch 落后则先调用fetchLogFromCMS主动拉取日志再比较。超时抛出带明细的AssertionError// 等待所有节点收敛 ClusterUtils.waitForCMSToQuiesce(cluster); // 等待指定 epoch Epoch targetEpoch ClusterUtils.maxEpoch(cluster); ClusterUtils.waitForCMSToQuiesce(cluster, targetEpoch);实际测试中AlterTopologyTest.java 即在拓扑变更后调用waitForCMSToQuiesce(cluster, cmsInstance)等待元数据应用完成。5.3 fetchLogFromCMS() / snapshotClusterMetadata()Epoch fetchLogFromCMS(IInvokableInstance inst, Epoch awaitedEpoch) Epoch fetchLogFromCMS(IInvokableInstance inst, long awaitedEpoch) Epoch snapshotClusterMetadata(IInvokableInstance inst)fetchLogFromCMS通过ClusterMetadataService.instance().fetchLogFromCMS(Epoch)从 CMS 拉取日志直到目标 epochL780-L789snapshotClusterMetadata触发triggerSnapshot()并返回快照 epochL801-L807。5.4 分布式观测MapString, Epoch getPeerEpochs(IInvokableInstance requester) // 向所有 peer 发送 TCM_CURRENT_EPOCH_REQ 汇总各节点 epoch SetString getCMSMembers(IInvokableInstance inst) // fullCMSMembers() 的地址集合getPeerEpochsL809-L829是少见的跨节点主动请求型工具向目录中所有地址发送Verb.TCM_CURRENT_EPOCH_REQ消息并通过 latch 等待全部回包返回peer - Epoch映射。5.5 调试字符串ListString getPeerDirectoryDebugStrings(IInvokableInstance inst) // directory.toDebugString() 按行拆分 ListString getTokenMapDebugStrings(IInvokableInstance inst) // tokenMap.toDebugString() 按行拆分 void logTokenMapDebugString(IInvokableInstance inst) // tokenMap.logDebugString() MapString, List[] getDataPlacementDebugInfo(IInvokableInstance inst) void logDataPlacementDebugString(IInvokableInstance inst, boolean byEndpoint)getDataPlacementDebugInfoL433-L456按 keyspace 返回二元数组下标 0 为读副本列表、下标 1 为写副本列表ReplicaGroups.toReplicaStringList()适合校验数据放置策略。6. CMS 暂停与控制CMS Pausing Control这是ClusterUtils最有价值的故障注入能力通过TestProcessorCMS 提交侧与TestChangeListener节点应用侧在精确时刻定格集群元数据流转构造竞态窗口。6.1 pauseBeforeCommit()提交前暂停CallableEpoch pauseBeforeCommit(IInvokableInstance cmsInstance, SerializablePredicateTransformation predicate)在 CMS 上注册谓词当待提交的Transformation匹配时暂停提交。返回的CallableEpoch会阻塞等待暂停发生并返回暂停时的 epoch。实现L602-L622通过((SwitchableProcessor) ClusterMetadataService.instance().processor()).delegate()取到TestProcessor并调用pauseIf(predicate, ...)超时 30 秒。CallableEpoch pauseHandle ClusterUtils.pauseBeforeCommit( cmsNode, transformation - transformation instanceof AddNode ); // 阻塞直到谓词匹配 Epoch pausedAt pauseHandle.call(); // 暂停期间执行其他操作 // ... // 恢复提交 ClusterUtils.unpauseCommits(cmsNode);6.2 getSequenceAfterCommit()提交后取 epochCallableEpoch getSequenceAfterCommit(IInvokableInstance cmsInstance, SerializableBiPredicateTransformation, Commit.Result predicate)在 CMS 提交侧注册提交谓词当某个 transformation 的提交结果匹配时返回其成功提交后的最新 epochresult.success().logState.latestEpoch()同样 30 秒超时L624-L653。6.3 应用侧暂停pauseBeforeEnacting / pauseAfterEnactingCallableVoid pauseBeforeEnacting(IInvokableInstance instance, long epoch) CallableVoid pauseBeforeEnacting(IInvokableInstance instance, Epoch epoch) CallableVoid pauseAfterEnacting(IInvokableInstance instance, Epoch epoch)在普通节点上暂停应用enact某个 epoch 的前/后时刻L529-L600。实现依赖TestChangeListener.instance通过AsyncPromise通知等待方pauseBeforeEnacting默认超时 30 秒pauseAfterEnacting默认 10 秒CallableVoid pauseHandle ClusterUtils.pauseBeforeEnacting(instance, targetEpoch); pauseHandle.call(); // 等待暂停 // 暂停期间执行操作 // ... ClusterUtils.unpauseEnactment(instance); // 恢复6.4 恢复与清理void unpauseCommits(IInvokableInstance instance) // 恢复 CMS 提交若实例已关闭则静默跳过 void unpauseEnactment(IInvokableInstance instance) // 恢复 epoch 应用 void clearAndUnpause(IInvokableInstance instance) // 清除全部暂停条件并恢复unpauseCommitsL655-L663对已关闭实例做了防护性判断避免在节点宕机后误操作。6.5 日志条目过滤void dropAllEntriesBeginningAt(IInvokableInstance instance, Epoch epoch) void clearEntryFilters(IInvokableInstance instance)dropAllEntriesBeginningAt为 CMS 日志添加过滤器丢弃epoch 指定值的所有条目L519-L522用于模拟元数据日志丢失clearEntryFilters清除全部过滤器。7. Ring 与 Gossip 监控Ring Gossip Monitoring7.1 ring()解析 nodetool ring 输出ListRingInstanceDetails ring(IInstance inst)内部执行inst.nodetoolResult(ring)并断言成功再用正则解析输出L925-L930。RingInstanceDetails字段address节点地址rack机架名statusUp / DownstateNormal / Leaving / Joining / Movingtokentoken 值7.2 Ring 断言立即检查ListRingInstanceDetails assertInRing(IInstance instance, IInstance expectedInRing) ListRingInstanceDetails assertNotInRing(IInstance instance, IInstance expectedInRing) ListRingInstanceDetails assertRingState(IInstance instance, IInstance expectedInRing, String state) ListRingInstanceDetails assertRingIs(IInstance instance, IInstance... expectedInRing) ListRingInstanceDetails assertRingIs(IInstance instance, Collection? extends IInstance expectedInRing) ListRingInstanceDetails assertRingIs(IInstance instance, SetString expectedRingAddresses)assertRingIs校验 ring 中恰好包含期望实例集合源码注释特别说明查询源节点自身可能不在 ring 中因此仅依赖期望集合比对地址使用 AssertJ 的isEqualTo保证严格相等L1134-L1142。7.3 Ring 等待轮询ListRingInstanceDetails awaitRingJoin(IInstance instance, IInstance expectedInRing) ListRingInstanceDetails awaitRingJoin(IInstance instance, String expectedInRing) void awaitRingJoin(Cluster cluster, int[] nodes, IInvokableInstance expectedInRing) ListRingInstanceDetails awaitRingHealthy(IInstance src) ListRingInstanceDetails awaitRingStatus(IInstance instance, IInstance expectedInRing, String status) ListRingInstanceDetails awaitRingState(IInstance instance, IInstance expectedInRing, String state)底层awaitRingL985-L998最多轮询 100 次、每次间隔 1 秒超时抛出带当前 ring 快照的AssertionError。awaitRingJoin判定条件为statusUp stateNormalawaitRingHealthy要求 ring 中所有实例均 Up 且 Normal。7.4 Gossip 信息MapString, MapString, String gossipInfo(IInstance inst) void assertGossipInfo(IInstance instance, InetSocketAddress expectedInGossip, int expectedGeneration, int expectedHeartbeat) MapString, MapString, String awaitGossipStatus(IInstance instance, IInstance expectedInGossip, String targetStatus) void awaitGossipSchemaMatch(ICluster? extends IInstance cluster) void awaitGossipSchemaMatch(IInstance instance) void awaitGossipStateMatch(ICluster? extends IInstance cluster, IInstance expectedInGossip, ApplicationState key)gossipInfo执行nodetool gossipinfo并解析为地址 - (generation, heartbeat, STATUS, ...)映射L1271-L1319。awaitGossipStatus兼容STATUS_WITH_PORT与旧版STATUS两个键使用 contains 匹配如传NORMAL。awaitGossipSchemaMatch会跳过未入环节点比较已入环节点的SCHEMA值是否一致若集群未启用Feature.GOSSIP则直接返回L1213-L1247。8. 系统表System Tablesvoid awaitInPeers(Cluster cluster, int[] nodes, IInstance expectedInPeers) void awaitInPeers(IInstance instance, IInstance expectedInPeers) boolean isInPeers(IInstance instance, IInstance expectedInPeers)isInPeersL1679-L1689通过内部查询SELECT tokens, data_center, rack FROM system.peers WHERE peer?判断目标节点是否已以完整信息出现在 peers 表中tokens 非空、dc/rack 非空。awaitInPeers轮询该条件最多 100 秒集群级重载会跳过自身节点节点不会出现在自己的 peers 中。9. 目录与文件Directories Files9.1 目录获取ListFile getDataDirectories(IInstance instance) File getCommitLogDirectory(IInstance instance) File getHintsDirectory(IInstance instance) File getSavedCachesDirectory(IInstance instance) File getJournalDirectory(IInstance instance) // Accord journal 目录配置项 accord.journal_directory File getCdcRawDirectory(IInstance instance) ListFile getDirectories(IInstance instance) // data commitlog hints saved caches journal cdc实现直接从IInstanceConfig读取对应配置项如data_file_directories、commitlog_directory、accord.journal_directory、cdc_raw_directory等源码注释坦承这依赖InstanceConfig的具体实现L1355-L1442。9.2 cleanup()重置实例到干净状态void cleanup(IInvokableInstance inst)依次执行stopUnchecked→ 删除全部可写目录递归→startup()L1449-L1458。等价于完全重来适用于需要清空数据重跑的测试。10. 节点操作Node Operations10.1 decommission()boolean decommission(IInvokableInstance leaving)在目标节点内部调用StorageService.instance.decommission(true)返回是否成功L856-L870boolean success ClusterUtils.decommission(cluster.get(3)); Assert.assertTrue(Decommission failed, success); ClusterUtils.assertNotInRing(cluster.get(1), cluster.get(3));10.2 getNodeId() / cancelInProgressSequences()NodeId getNodeId(IInvokableInstance target) int getNodeId(IInvokableInstance target, IInvokableInstance executor) boolean cancelInProgressSequences(IInvokableInstance executor) boolean cancelInProgressSequences(NodeId nodeId, IInvokableInstance executor)getNodeId通过executor节点查询ClusterMetadata.current().directory.peerId(target地址)得到 TCM 视角的NodeIdL872-L891。cancelInProgressSequences调用StorageService.instance.cancelInProgressSequences(...)用于清理节点加入/离开过程中残留的进行中序列。10.3 mode() / assertModeJoined() / isMigrating()StorageService.Mode mode(IInvokableInstance inst) void assertModeJoined(IInvokableInstance inst) boolean isMigrating(IInvokableInstance instance)mode返回StorageService.Mode可取STARTING、NORMAL、JOINING、LEAVING、DECOMMISSIONED、MOVING、DRAINING、DRAINEDL1691-L1695assertModeJoined断言模式为NORMALisMigrating查询ClusterMetadataService.instance().isMigrating()判断 CMS 是否正在迁移L675-L678。11. 日志监控Log Monitoringvoid runAndWaitForLogs(Runnable r, String waitString, AbstractClusterI cluster) void runAndWaitForLogs(Runnable r, String waitString, IInstance... instances)先在每个目标实例上logs().mark()标记位置执行操作再逐个watchFor(mark, waitString)等待日志出现L484-L497ClusterUtils.runAndWaitForLogs( () - cluster.get(2).nodetool(bootstrap), Bootstrap completed, cluster.get(1), cluster.get(2), cluster.get(3) );12. 地址管理Address Managementvoid updateAddress(IInstance instance, String address) String getBroadcastAddressHostWithPortString(IInstance target) InetSocketAddress getNativeInetSocketAddress(IInstance target) int getIntConfig(IInstanceConfig config, String configName, int defaultValue)updateAddressL1487-L1530同时修改broadcast_address、listen_address、broadcast_rpc_address、rpc_address四个配置项只能在该实例处于关闭状态时调用否则行为未定义。实现细节包括强制清空InstanceConfig缓存的InetSocketAddress - InetAddressAndPort映射否则启动会忽略新配置并通过反射更新NetworkTopology内部的 HashMap源码 TODO 提示 NetworkTopology 不支持动态增删且非线程安全节点运行期间变更存在风险。getNativeInetSocketAddress返回 CQL 原生协议地址端口读取native_transport_port缺省回退 9042L1552-L1556getIntConfig在配置缺失NPE时返回默认值L1575-L1585。13. Accord 测试辅助13.1 事务状态查询T extends IInstance LinkedHashMapString, SimpleQueryResult queryTxnState( AbstractClusterT cluster, TxnId txnId, int... nodes) T extends IInstance String queryTxnStateAsString(AbstractClusterT cluster, TxnId txnId, int... nodes) T extends IInstance void queryTxnStateAsString(StringBuilder sb, AbstractClusterT cluster, TxnId txnId, int... nodes)queryTxnStateL1704-L1717对每个目标节点执行SELECT * FROM system_accord.TXN_BLOCKED_BY WHERE txn_id?SchemaConstants.VIRTUAL_ACCORD_DEBUGAccordDebugKeyspace.TXN_BLOCKED_BY跳过已关闭节点nodes为空时遍历整个集群。queryTxnStateAsString将结果拼为可读文本AccordTestBase.java 中即用它输出故障现场。13.2 tableId() / awaitAccordEpochReady()TableId tableId(Cluster cluster, String ks, String table) void awaitAccordEpochReady(Cluster cluster, long epoch)tableId通过Schema.instance.getKeyspaceInstance(ks).getColumnFamilyStore(table).getTableId()获取表 IDL1743-L1747awaitAccordEpochReady等待所有存活实例上AccordService.instance().epochReady(Epoch, EpochReady::reads)完成即指定 epoch 的读路径就绪L1749-L1764。14. 工具类Utility Classes14.1 Rangepublic static class Range implements Serializable { public final String left, right; public Range(String left, String right) public Range(long left, long right) public long left() public long right() }表示一个 token 范围标记为Shared可跨类加载器序列化。getPrimaryRanges返回的即为此类实例。14.2 RingInstanceDetailspublic static final class RingInstanceDetails { private final String address; // 节点地址 private final String rack; // 机架名 private final String status; // Up/Down private final String state; // Normal/Leaving/Joining/Moving private final String token; // token 值 public String getAddress() / getRack() / getStatus() / getState() / getToken() }ring()/awaitRing*/assert*系列返回的解析结果对象实现完整的equals/hashCode/toStringL1587-L1652可直接用于断言比较。15. 实战组合一个完整的节点替换测试综合以上 API一个典型的替换故障节点测试流程如下模式见 SKILL.md 及 HostReplacementTest.java 等用例try (Cluster cluster Cluster.build(3).start()) { // 1. 创建 keyspace/table省略 // 2. 模拟节点 2 被 kill -9 IInvokableInstance original cluster.get(2); ClusterUtils.stopAbrupt(cluster, original); // 3. 创建并启动替换节点 IInvokableInstance replacement ClusterUtils.replaceHostAndStart( cluster, original, (inst, properties) - properties.set(RING_DELAY, 5000)); // 4. 等待元数据收敛CMS quiesce ClusterUtils.waitForCMSToQuiesce(cluster); // 5. 等待替换节点入环 ClusterUtils.awaitRingJoin(cluster.get(1), replacement); ClusterUtils.awaitRingHealthy(cluster.get(1)); // 6. 校验 ring 成员关系 ClusterUtils.assertInRing(cluster.get(1), replacement); ClusterUtils.assertNotInRing(cluster.get(1), original); }16. 最佳实践小结只在测试侧调用ClusterUtils是Isolated类禁止在节点内部runOnInstance的 lambda 内调用所有内部逻辑走callOnInstance/runOnInstance。善用等待而非 sleep优先使用awaitRingJoin、awaitRingHealthy、waitForCMSToQuiesce、awaitGossipSchemaMatch等带轮询超时的方法它们会在超时时给出包含现场快照的AssertionError便于定位。区分断言与等待assert*系列立即检查不满足即失败await*系列轮询等待默认约 100 秒、1 秒间隔二者应组合使用。CMS 暂停是竞态利器pauseBeforeCommit/pauseBeforeEnacting配合unpauseCommits/unpauseEnactment可在拓扑变更的精确时机注入竞争使用后务必恢复或调用clearAndUnpause。注意限制getTokens仅适用于显式配置 token 的实例updateAddress仅能在节点关闭时调用addInstance要求集群创建时就规划好拓扑。替换/退役后必须收敛replaceHostAndStart/decommission之后用waitForCMSToQuiesceawaitRingJoin/assertNotInRing完成双维度校验元数据 epoch 与 ring 视图。通过将本指南与 SKILL.md、advanced_patterns.md、classloader_guide.md 配合使用你可以在单 JVM 内高效复现多节点集群的各类拓扑与故障场景。【免费下载链接】cassandraOpen source transactional distributed database. Linear scalability and proven fault-tolerance on commodity hardware or cloud infrastructure without compromising performance.项目地址: https://gitcode.com/GitHub_Trending/cassa/cassandra创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考