所以我用这个例子嵌入Kafka和这个
我稍微修改了这个示例,并用一些数据库(比如h2db)更新了kafka侦听器。
现在在我的单元测试中,当我想检查数据是否在db中可用时,我得到的是null。另外,我不知道如何检查数据库手动作为h2是一个内存基数据库。
下面是更新的部分:在receiver类中
@Autowired
DataTableRepository repository;
@KafkaListener(topics = "${kafkatest.topic}")
public void receive(ConsumerRecord<String, DataTable> consumerRecord) {
LOGGER.info("received payload='{}'", consumerRecord.toString());
repository.save(consumerRecord.value());
latch.countDown();
}
单元测试:
@Autowired
DataTableRepository repository;
@Test
public void testReceive() throws Exception {
DataTable table = new DataTable(1, "Sending with default template");
template.send(topic, table);
receiver.getLatch().await(10000, TimeUnit.MILLISECONDS);
DataTable dt = repository.getOne(table.getId());
assertNotNull(dt);
assertThat(receiver.getLatch().getCount(), equalTo(0L));
}
但是dt总是为空。我也不能检查数据库也,因为它得到停止后,测试停止。有人知道怎么做吗?
暂无答案!
目前还没有任何答案,快来回答吧!