什么可以用作CassandraWriterBolt的测试存根?

Question

我从Kafka读取了一个json，FieldExtractionBolt读取json将数据提取为元组值并传递给CassandraWriterBolt，CassandraWriterBolt又在Cassandra中写入一条记录，将所有这些元组值写入单独的列。

卡夫卡上的 JSON 消息 -

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

FieldExtractionBolt -

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")));

CassandraWriterBolt -

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

我尝试根据此处给出的答案编写测试 - 如何通过以编程方式插入消息来测试 Storm 拓扑的 E2E 功能

在我的项目中，我在 Spring 配置中定义了我所有的螺栓、喷口和流。这使得编写/读取我的拓扑非常容易。我通过从 ApplicationContext 获取螺栓、喷口和流 bean 来构建拓扑。在我的 Spring 配置中，KafkaSpout 和 CassandraWriterBolt 被定义为 'prod' profile，因此它们只用于 prod，而在 'test' profile， 我为 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只是将第一个值读取为元组(仅货币对并跳过买入价和卖出价)。似乎TestWordCounter在这里不是一个正确的选择。CassandraWriterBolt 的正确存根是什么，以便我可以断言它将收到 2 条记录，一条用于英镑兑日元，另一条用于英镑兑美元以及他们的买入价和卖出价？

Answer 1

Testing.readTuples(results, "cassandraWriterBolt")将返回"cassandraWriterBolt"发出的元组。这就是您要测试的内容吗？我认为您试图断言"cassandraWriterBolt"接收哪些元组，而不是它发出的内容。

你可以在这里做两件事。您可以使用readTuples从发射到 Cassandra 螺栓的螺栓中读取，而不是从 Cassandra 螺栓读取。如果您的拓扑结构很简单(例如，写入Cassandra bolt的不同螺栓不多)，这是一个不错的解决方案。

更好的解决方案(IMO)是编写一个简单的短截螺栓来替换TestWordCounter。bolt 唯一应该做的是接收输入元组，确认它，然后在新元组中发出值。

execute(Tuple input, BasicOutputCollector collector) {
collector.emit(input.getValues());
}

然后，您可以使用readTuples读取 bolt 发出的元组，这些元组将与它收到的值相同。