SpringBoot使用Testcontainers+Avro消息测试Kafka消费者

您有一个 Spring Boot 微服务来管理用户数据。该微服务侦听来自 Kafka 的传入事件（例如用户创建、更新或删除），将它们转换为您自己的业务对象，将它们写入 PostgreSQL 数据库并通过 REST 接口将它们提供给您的前端。整个基础设施提供Avro消息和 Confluent模式注册表。
现在您要测试您的 Kafka 消费者是否读取了事件，将它们转换为您的数据库实体并保存它们。
多种解决方案：使用MockSchemaRegistryClient，或编写自己的自定义的Avro去/串行化器，或设置一个Testcontainers生态系统与Kafka，一个 Zookeeper 和一个 Confluent Schema Registry，或者在 spring-kafka-test 依赖中使用Spring提供的 EmbeddedKafka 。
所有这些解决方案都有其有效的优点和缺点。
在这篇博文中，我将提出一个解决方案，它使用最少的测试容器集来提供控制、速度和效率之间的最佳折衷。我上面列出的其他解决方案各有利弊，我将在本文后面介绍。

设置
让我引导您完成较小的设置，模拟上述情况。我们有一个 Spring Boot 应用程序、一个 PostgreSQL 数据库和我们的 Kafka 消费者。应用程序需要为添加或修改的用户监听 Kafka 消息，并且必须相应地更新数据库项。您可以在此处找到完整的代码存储库。我添加了 UserEvent 类仅用于编译目的，通常这会通过 Avro 插件最终出现在您生成的类中。

让我们解决第一个问题：如何在我们的测试环境中与 Schema Registry 交互？我们需要一个模拟Mock。隐藏在 Confluent 的模式注册表包中，在AbstractKafkaAvroSerDeConfig类中，您可以找到模式注册表 url 的注释：
可用于注册或查找模式的模式注册表实例的以逗号分隔的 URL 列表。如果您希望连接到模拟模式注册表以进行测试，您可以使用“mock://”伪协议指定范围。例如，“mock://my-scope-name”对应于“MockSchemaRegistry.getClientForScope(“my-scope-name”)”。
所以这意味着我们可以用一个虚构的模式注册表 URL 配置 Kafka 生产者和消费者，只需要以“mock://”开头，你就会自动开始使用MockSchemaRegistryClient。这样您就不需要显式启动MockSchemaRegistryClient并相应地配置所有内容。
这也消除了对 Confluent Schema Registry Container 的需求，还有运行带有嵌入式 Zookeeper的Kafka Testcontainer，我们不再需要额外的 Zookeeper 容器，我们只需要一个 Testcontainer 来进行消息传递。
这样我最终只有两个测试容器：Kafka 和数据库。
设置和配置您的容器：

 protected static final PostgreSQLContainer<?> postgreSQLContainer =
      new PostgreSQLContainer<>("postgres:10.9")
          .withPassword("postgres")
          .withUsername("postgres")
          .withExposedPorts(5432)
          .withReuse(true);

  protected static final KafkaContainer kafkaContainer =
      new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.3"))
          .withEmbeddedZookeeper()
          .withEnv("KAFKA_LISTENERS", "PLAINTEXT://0.0.0.0:9093 ,BROKER://0.0.0.0:9092")
          .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT")
          .withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER")
          .withEnv("KAFKA_BROKER_ID", "1")
          .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
          .withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1")
          .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
          .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
          .withEnv("KAFKA_LOG_FLUSH_INTERVAL_MESSAGES", Long.MAX_VALUE + "")
          .withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0")
          .withNetwork(network);

  static {
    Startables.deepStart(Stream.of(postgreSQLContainer, kafkaContainer)).join();
  }

并使用他们的测试配置创建 Kafka 生产者和消费者：

 public static KafkaProducer<Object, Object> createProducer() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
    props.put(ProducerConfig.ACKS_CONFIG, "all");
    props.put(ProducerConfig.RETRIES_CONFIG, 0);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(
        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        io.confluent.kafka.serializers.KafkaAvroSerializer.class);
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "mock://testUrl");
    props.put(ProducerConfig.CLIENT_ID_CONFIG, "kafkatest");
    return new KafkaProducer<>(props);
  }

 public static KafkaConsumer<String, Object> createEventConsumer() {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
    props.put(
        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "mock://testUrl");
    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "kafkatest");
    return new KafkaConsumer<>(props);
  }

这里我们只添加了模拟前缀模式注册表 url，没有特殊的序列化器或反序列化器。不要忘记在 Kafka 消费者配置中将 specific.avro.reader 的属性设置为“true”，以确保消费者不会因类转换异常而失败。
现在，在您的测试中，您创建您希望 Kafka 生产者发送的事件，让您的 Kafka 消费者订阅该主题，并从KafkaTestUtils 中获取 ConsumerRecord以供您自己的侦听器处理。然后你测试你的听众所做的任何事情的结果,就这样。

利弊？
当然，运行时间是一个问题在我的机器上运行平均需要 5 秒。我使用一个 Kafka 和一个 Postgres 容器的版本平均需要 15 秒才能运行。
现在想象一下使用 4 个测试容器（Schema Registry、Kafka 和 Zookeeper，以及数据库容器）可以运行多长时间。并且有更多的代码，需要测试更复杂的业务逻辑。
测试容器在完全可用之前需要启动和联网。我的设置的一个更快的替代方案是将所有基础设施移动到内存中，使用 EmbeddedKafka 作为代理并用 H2 数据库替换 PostgreSQL。您可以在“embeddedKafka”分支中找到EmbeddedKafka的工作版本（我的命名技巧非常有想象力）。
但是使用 EmbeddedKafka 需要注意依赖版本。Confluent Schema Registry Client 带有自己的 Zookeeper 版本，根据 Schema Registry Client 的版本，您最终可能会得到与 kafka-test 依赖项预期的版本不同的 Zookeeper 版本，这将导致 ClassNotFound 异常运行测试时。有关更多详细信息，请查看 Confluent 的版本矩阵和Spring的Kafka 客户端列表。
因此，由于版本限制，我坚持使用容器设置。对于这个演示应用程序，它根本不重要，但为了避免意外行为，我建议尽可能接近生产版本。当然，使用 H2 也可能并不总是一种选择，这取决于您使用生产数据库的功能（例如约束、json 数据类型、某些连接语句）。所以这是关于控制和测试运行时的妥协。
另外，我是否已经提到“视情况而定”？
你可以在github上找到代码，主分支是两个容器的演示，embeddedKafka分支是容器免费版本。