使用 poll() 的消费者单元测试永远不会收到任何东西

Question

考虑以下代码：

@Test(singleThreaded = true)

public class KafkaConsumerTest

{

  private KafkaTemplate<String, byte[]> template;

  private DefaultKafkaConsumerFactory<String, byte[]> consumerFactory;

  private static final KafkaEmbedded EMBEDDED_KAFKA;

  static {

      EMBEDDED_KAFKA = new KafkaEmbedded(1, true, "topic");

      try { EMBEDDED_KAFKA.before(); } catch (final Exception e) { e.printStackTrace(); }

    }

  @BeforeMethod

  public void setUp() throws Exception {

    final Map<String, Object> senderProps = KafkaTestUtils.senderProps(EMBEDDED_KAFKA.getBrokersAsString());

    senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

    senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);

    final ProducerFactory<String, byte[]> pf = new DefaultKafkaProducerFactory<>(senderProps);

    this.template = new KafkaTemplate<>(pf);

    this.template.setDefaultTopic("topic");

    final Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("sender", "false", EMBEDDED_KAFKA);

    this.consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);

    this.consumerFactory.setValueDeserializer(new ByteArrayDeserializer());

    this.consumerFactory.setKeyDeserializer(new StringDeserializer());

  }

我正在尝试向 a 发送消息KafkaTemplate并使用Consumer.poll(). 我使用的测试框架是TestNG。

发送作品，我已经验证使用我在网上找到的“常用”代码（在 上注册一个消息侦听器KafkaMessageListenerContainer）。

只是，我从来没有在消费者那里收到过任何东西。我已经针对“真实”的 Kafka 安装尝试了相同的序列 (create Consumer, poll())，并且它有效。

因此，我设置ConsumerFactory? 任何帮助将不胜感激！