我从kafka读取了一个json，fieldextractionbolt读取到json将数据提取成元组值并传递给cassandrawriterbolt，然后cassandrawriterbolt在cassandra中写入一条记录，将所有元组值写入单独的列。
Kafka上的json消息-

{"pair":"GBPJPY","bid":134.4563,"ask":134.4354}

现场拔出螺栓-

String message = tuple.getStringByField("message");
Map values = new Gson().fromJson(message, Map.class);
basicOutputCollector.emit(new Values(values.get("pair"), values.get("bid"), values.get("ask")));

Cassandra维特博尔特-

return (CassandraWriterBolt) new CassandraWriterBolt(async(simpleQuery("INSERT INTO currency(pair, ask, bid) VALUES (?, ?, ?);").with(fields("pair", "ask", "bid")))

我试着根据这里给出的答案编写一个测试——如何通过编程插入消息来测试storm拓扑的功能
在我的项目中，我在spring配置中定义了所有螺栓、喷口和流。这使得写/读我的拓扑非常容易。我通过从applicationcontext获取bolt、spouts和streambean来构建拓扑。在我的spring配置中，kafkaspout和cassandrawriterbolt是在prod配置文件下定义的，因此它们只能在prod中使用，而在test配置文件下，我为kafkaspout和cassandrawriterbolt定义存根。对于kafkaspout，我使用fixedtouplespout，对于cassandrawriterbolt，我使用testwordcounter。
这是我的测试

@Test
        public void testTopology(){
        StormTopology topology = SpringBasedTopologyBuilder.getInstance().buildStormTopologyUsingApplicationContext(applicationContext);
        TestJob COMPLETE_TOPOLOGY_TESTJOB = (cluster) -> {
              MockedSources mocked = new MockedSources();
                    mocked.addMockData("kafkaSpout",
                new Values("{\"pair\":\"GBPJPY\",\"bid\":134.4563,\"ask\":134.4354}"),
        new Values("{\"pair\":\"GBPUSD\",\"bid\":1.4563,\"ask\":1.4354}"));

        Config topoConf = new Config();
        topoConf.setNumWorkers(2);

        CompleteTopologyParam ctp = new CompleteTopologyParam();
        ctp.setMockedSources(mocked);
        ctp.setStormConf(topoConf);
        Map<String, List<FixedTuple>> results = Testing.completeTopology(cluster, topology, ctp);
                    List<List<Object>> cassandraTuples = Testing.readTuples(results, "cassandraWriterBolt");
        List<List<Object>> expectedCassandraTuples = Arrays.asList(Arrays.asList("GBPJPY", 1), Arrays.asList("GBPUSD", 1),
                Arrays.asList("134.4563", 1), Arrays.asList("1.4563", 1), Arrays.asList("134.4354", 2));
        assertTrue(expectedCassandraTuples + " expected, but found " + cassandraTuples,
                Testing.multiseteq(expectedCassandraTuples, cassandraTuples));
    MkClusterParam param = new MkClusterParam();
    param.setSupervisors(4);

    Testing.withSimulatedTimeLocalCluster(param, COMPLETE_TOPOLOGY_TESTJOB);
}

@Configuration
@Import(MainApplication.class)
public static class TestConfig
{
    @Bean
    public IRichSpout kafkaSpout(){
        return new FixedTupleSpout(Arrays.asList(new FixedTuple(Arrays.asList("{\"pair\":\"GBPJPY\",\"bid\":134.4563,\"ask\":134.4354"))), new Fields(new String[]{"message"}));
    }

    @Bean
    public IBasicBolt cassandraWriterBolt(){
        return new TestWordCounter();
    }
}

我得到的结果不是我所期望的。我有以下错误-

java.lang.AssertionError: [[GBPJPY, 1], [GBPUSD, 1], [134.4563, 1], [1.4563, 1], [134.4354, 2]] expected, but found [[GBPJPY, 1], [GBPUSD, 1]]

看起来，testwordcounter只是将第一个值作为元组读取（仅限货币对，跳过bid和ask）。似乎testwordcounter不是一个正确的选择。什么是正确的存根为cassandrawriterbolt，以便我可以Assert，它将收到两个记录英镑日元和英镑美元的另一个与他们的出价和要价以及？