为 Kafka Consumer 编写 JUnit 测试答案

【问题标题】：Writing JUnit tests for Kafka Consumer为 Kafka Consumer 编写 JUnit 测试
【发布时间】：2018-11-27 23:36:59
【问题描述】：

我有一个订阅某个主题的 kafka 消费者。实施工作正常。但是当试图为此实现单元测试时，由于它是由Runnable 接口实现的，所以会出现问题。

实施

@Override
public void run() {
    kafkaConsumer.subscribe(kafkaTopics);

    while (true) {
        ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);
        Map<String, InventoryStock> skuMap = new LinkedHashMap<>();

        try {
            // populating sku map with consumer record
            for (ConsumerRecord<String, String> record : records) {
                populateMap(skuMap, record.value());
            }

            if (MapUtils.isNotEmpty(skuMap)) {
                // writing sku inventory with populated sku map
                inventoryDao.updateInventoryTable(INVENTORY_JOB_ID, skuMap);
            }
        } catch (Exception e) {

        }
        kafkaConsumer.commitAsync();
    }
}

我尝试使用MockConsumer 实现测试。但在实现时需要分配给消费者。但是实施中的消费者并没有暴露在外面。这是我尝试过的。

@Before
public void onBefore() {
    MockitoAnnotations.initMocks(this);

    Properties consumerProps = new Properties();
    consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
    consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
    skuInventoryConsumer = new SkuInventoryConsumer(consumerProps);

    KafkaConsumer kafkaConsumerMock = mock(KafkaConsumer.class);

    Whitebox.setInternalState(skuInventoryConsumer, "LOGGER", LOGGER);
    Whitebox.setInternalState(skuInventoryConsumer, "kafkaConsumer", kafkaConsumerMock);

}

@Test
public void should_subscribe_on_topic() {
    consumer.assign(Arrays.asList(new TopicPartition("my_topic", 0)));

    HashMap<TopicPartition, Long> beginningOffsets = new HashMap<>();
    beginningOffsets.put(new TopicPartition("my_topic", 0), 0L);
    consumer.updateBeginningOffsets(beginningOffsets);

    consumer.addRecord(new ConsumerRecord<>("my_topic", 0, 0L, "mykey", "myvalue0"));
    consumer.addRecord(new ConsumerRecord<>("my_topic", 0, 1L, "mykey", "myvalue1"));
    consumer.addRecord(new ConsumerRecord<>("my_topic", 0, 2L, "mykey", "myvalue2"));
    consumer.addRecord(new ConsumerRecord<>("my_topic", 0, 3L, "mykey", "myvalue3"));
    consumer.addRecord(new ConsumerRecord<>("my_topic", 0, 4L, "mykey", "myvalue4"));
}

因为它是runnable 并且消费者没有暴露，所以这个测试没有按预期工作。我该如何解决这个问题？

【问题讨论】：

你想测试什么？也许您可以添加倒计时闩锁以在消费者消费时获取回调（而不是多线程方式的模拟方式）
你找到如何测试消费者了吗？
不。找不到测试由Runnable 实现的消费者的方法。但是找到了一些方法来测试在普通类中实现的普通消费者。
你能分享给我们一个链接吗？
我现在没有。给我一些时间。我会给你一个链接。

标签： java apache-kafka

【解决方案1】：

我建议使用 Mockito，如下例所示

Consumer<String, String> kafkaConsumerLocal = mock(Consumer.class);
KafkaConsumer kafkaConsumer = spy(new KafkaConsumer("topic-name"));

ReflectionTestUtils.setField(kafkaConsumer, "threadPoolCount", 1);
ReflectionTestUtils.setField(kafkaConsumer, "consumer", kafkaConsumerLocal);

doNothing().when(kafkaConsumer).runConsumer();
doNothing().when(kafkaConsumer).addShutDownHook();
doReturn(kafkaConsumerLocal).when(consumerInit).getKafkaConsumer();

【讨论】：