@EmbeddedKafka 不适用于 Spring-Boot 3.2* 或 Spring 6.*

Question

我们已将 spring-boot 版本升级为 3.2.1。发布嵌入式 Kafka(@EmbeddedKafka) 无法启动。

Java版本：17.0.8 MVN版本：3.5.8

子模块POM

<modelVersion>4.0.0</modelVersion>
<parent>
    <groupId>com.fault</groupId>
    <artifactId>topology</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</parent>

<name>cdc-service</name>
<artifactId>cdc-service</artifactId>
<packaging>jar</packaging>

<properties>
    <java.version>17</java.version>       
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>com.github.ulisesbocchio</groupId>
        <artifactId>jasypt-spring-boot</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
   
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka-test</artifactId>
        <scope>test</scope>
    </dependency>
    <!-- <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
    </dependency>  -->
  
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <configuration>
                <mainClass>com.cdc.MyServiceApplication</mainClass>
            </configuration>
            <executions>
                <execution>
                    <goals>
                        <goal>repackage</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

父模块 Pom

<properties>
        <java.version>17</java.version>
        <revision>1.0.0-SNAPSHOT</revision>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <!-- common -->
        <spring.boot.version>3.2.1</spring.boot.version>
        <jasypt.spring.version>3.0.5</jasypt.spring.version>
</properties>

它抛出异常以下

java.lang.NoClassDefFoundError: org/apache/kafka/coordinator/group/assignor/RangeAssignor
    at kafka.server.Defaults$.<clinit>(KafkaConfig.scala:179) ~[kafka_2.13-3.6.0.jar:na]
    at kafka.server.KafkaConfig$.<clinit>(KafkaConfig.scala:746) ~[kafka_2.13-3.6.0.jar:na]
    at kafka.server.KafkaConfig.DeleteTopicEnableProp(KafkaConfig.scala)

我们添加了所需的 jar (kafka-cordinator jar) 来解决此问题。但在那之后，它会抛出一些其他类未找到的错误。是否需要这种依赖性？

测试类级别注释

@SpringBootTest
@DirtiesContext
@EmbeddedKafka(partitions = 1, brokerProperties = {"auto.create.topics.enable=false"}, topics = {"topic1","topic2"}, ports = 9092)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@TestInstance(TestInstance.Lifecycle.PER_CLASS)

最新版本这里需要添加什么额外的东西吗？

编辑2

使用 kraft = false 我们的实际问题没有解决。 @EmbeddedKafka 仍然无法工作。

在 Sample05 项目中，我们添加了一个新的测试类，如下所示，并设置 spring.kafka.global.embedded.enabled = false

  @SpringBootTest
@DirtiesContext
@EmbeddedKafka(partitions = 1, brokerProperties = {"auto.create.topics.enable=false"}, topics = {"topic1","topic2"}, ports = 9092, kraft  = false )
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class TestEmbeddedKF {

    @Autowired
    private EmbeddedKafkaBroker embeddedKafkaBroker;
    
    @BeforeAll
    public void init() {
    }
    }

现在在示例项目中抛出以下错误，我们在项目中遇到了同样的错误。

java.lang.NoClassDefFoundError: scala/collection/Iterable
        at kafka.utils.TestUtils.tempDir(TestUtils.scala)
        at org.springframework.kafka.test.EmbeddedKafkaZKBroker$EmbeddedZookeeper.<init>(EmbeddedKafkaZKBroker.java:800)
        at org.springframework.kafka.test.EmbeddedKafkaZKBroker.afterPropertiesSet(EmbeddedKafkaZKBroker.java:298)
        at org.springframework.kafka.test.context.EmbeddedKafkaContextCustomizer.customizeContext(EmbeddedKafkaContextCustomizer.java:130)
        at org.spr

在我们的模块中

java.lang.NoClassDefFoundError: scala/collection/Map
at kafka.utils.TestUtils.tempDir(TestUtils.scala) ~[kafka_2.13-3.6.1-test.jar:na]
at org.springframework.kafka.test.EmbeddedKafkaZKBroker$EmbeddedZookeeper.<init>(EmbeddedKafkaZKBroker.java:800) 

在这两种情况下，错误都来自同一文件 TestUtils.scala。这个问题有解决方案/缺陷吗？

使用动态端口选项，我们也遇到了相同的异常。

Answer 1

从 Spring for Apache Kafka 开始的

@EmbeddedKafka

默认使用 KRaft 算法。 Kafka 客户端的

3.1

不支持显式端口。 或者不要使用那个

KafkaClusterTestKit

或使用

ports = 9092

。在已关闭的问题中查看更多信息：

https://github.com/spring-projects/spring-kafka/issues/2916

更新

根据您最新的错误以及 POM 中显示的内容，您缺少测试范围的这些依赖项：

kraft = false