Apache Storm：在拓扑实例内单独测试Bolt的技术问询

你的思路完全靠谱！用自定义测试Spout替代Kafka Spout，在精简的Storm拓扑实例里测试Bolt的完整行为（包括和Cassandra的交互），这正是Storm生态里测试这类组件的标准玩法。我给你一步步拆解具体怎么实现：

1. 先搞定测试依赖 #

首先要添加Storm的测试工具包，要是用嵌入式Cassandra做真实数据验证的话，还得加cassandra-unit。以Maven为例，把这些放在test scope里：

<dependency>
    <groupId>org.apache.storm</groupId>
    <artifactId>storm-testing</artifactId>
    <version>${storm.version}</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.cassandraunit</groupId>
    <artifactId>cassandra-unit</artifactId>
    <version>${cassandraunit.version}</version>
    <scope>test</scope>
</dependency>

2. 写个测试用的Mock Spout #

这个Spout的核心就是模拟Kafka Spout输出的Tuple结构——字段名、数据类型要和生产环境完全一致，这样才能保证Bolt接收到的Tuple和真实场景一样。比如：

public class TestDataSpout extends BaseRichSpout {
    private SpoutOutputCollector collector;
    private final List<List<Object>> testTuples;

    // 构造时传入要发送的测试数据
    public TestDataSpout(List<List<Object>> testTuples) {
        this.testTuples = testTuples;
    }

    @Override
    public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void nextTuple() {
        // 发送所有测试数据后休眠，避免无限循环发送
        testTuples.forEach(tuple -> collector.emit(tuple));
        Utils.sleep(1000);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 必须和你的Kafka Spout输出字段完全匹配！
        declarer.declare(new Fields("user_id", "event_type", "timestamp"));
    }
}

3. 搭建测试拓扑并运行 #

接下来用LocalCluster（Storm的本地测试集群）启动一个只包含测试Spout和目标Bolt的拓扑，这样就能模拟真实的拓扑运行环境，同时跳过Kafka的依赖。示例代码如下：

@Test
public void testBusinessBoltInTopology() throws Exception {
    // 1. 准备测试数据，和业务Tuple结构对齐
    List<List<Object>> testData = Arrays.asList(
        Arrays.asList("user_001", "login", System.currentTimeMillis()),
        Arrays.asList("user_002", "logout", System.currentTimeMillis())
    );

    // 2. 初始化测试组件
    Spout testSpout = new TestDataSpout(testData);
    Bolt targetBolt = new YourBusinessCassandraBolt(); // 你的业务Bolt实例

    // 3. 构建测试拓扑，分组策略要和生产环境一致！
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("test-data-spout", testSpout);
    builder.setBolt("business-cassandra-bolt", targetBolt)
           .shuffleGrouping("test-data-spout"); // 生产用什么分组这里就用什么

    // 4. 配置本地集群并启动拓扑
    Config conf = new Config();
    conf.setDebug(false);
    conf.setNumWorkers(1); // 单worker足够测试

    try (LocalCluster cluster = new LocalCluster()) {
        cluster.submitTopology("bolt-test-topology", conf, builder.createTopology());
        // 等待Bolt处理完数据，时间根据你的业务处理速度调整
        Thread.sleep(5000);

        // 5. 核心步骤：验证Cassandra中的数据是否符合预期
        verifyCassandraWriteResults();
    }
}

4. 验证Cassandra的写入结果 #

这里有两种方案，选哪种看你的测试需求：

方案一：用嵌入式Cassandra做真实验证 #

这种方式最接近生产场景，用cassandra-unit启动一个内存中的Cassandra实例，提前创建好表结构，测试后直接查询验证：

private Session cassandraSession;

@Before
public void setupEmbeddedCassandra() throws Exception {
    // 启动嵌入式Cassandra，加载你的表结构脚本
    Cluster cluster = CassandraUnitRule.builder()
        .withClusterName("test-cluster")
        .withPort(9142)
        .withKeyspace("your_business_keyspace")
        .withSchemaScript("classpath:cassandra_schema.cql")
        .build()
        .getCluster();
    cassandraSession = cluster.connect("your_business_keyspace");
}

private void verifyCassandraWriteResults() {
    // 查询验证login事件是否写入
    ResultSet loginResult = cassandraSession.execute(
        "SELECT * FROM user_events WHERE user_id = ? AND event_type = ?", 
        "user_001", "login"
    );
    Assert.assertEquals(1, loginResult.all().size());

    // 验证logout事件
    ResultSet logoutResult = cassandraSession.execute(
        "SELECT * FROM user_events WHERE user_id = ? AND event_type = ?", 
        "user_002", "logout"
    );
    Assert.assertEquals(1, logoutResult.all().size());
}

@After
public void shutdownCassandra() {
    cassandraSession.close();
}

方案二：Mock Cassandra客户端（轻量快速） #

如果不想启动嵌入式数据库，也可以用Mockito等工具Mock Bolt里的Cassandra驱动，只验证Bolt是否正确调用了写入方法：

@Test
public void testBoltWithMockCassandra() throws Exception {
    // Mock Cassandra Session
    Session mockSession = Mockito.mock(Session.class);
    // 把Mock Session注入到Bolt里（需要你的Bolt支持构造函数/Setter注入）
    YourBusinessCassandraBolt bolt = new YourBusinessCassandraBolt(mockSession);

    // 构建测试拓扑并运行（前面的代码逻辑一致）
    // ...

    // 验证Bolt是否正确调用了Cassandra的写入方法
    Mockito.verify(mockSession, Mockito.times(2))
           .execute(Mockito.anyString()); // 验证调用次数，或者更精确的SQL语句
}

几个额外的小技巧 #

• 尽量保持测试拓扑和生产拓扑的一致性：比如分组策略、并行度配置，这样测试结果更可信。
• 用Storm的Testing工具类：org.apache.storm.testing.Testing里有很多便利方法，比如completeTopology可以自动处理拓扑生命周期，还能收集Bolt的输出Tuple做验证。
• 测试后清理数据：不管用嵌入式Cassandra还是Mock，测试完成后都要清理数据，避免影响下一次测试。

内容的提问来源于stack exchange，提问作者breezymri