如何结合@ServiceConnection与GenericContainer实现自定义ConnectionDetails

我在测试中用新的@ServiceConnection注解管理KafkaContainer，效果符合预期——自动配置时会用KafkaContainerConnectionDetails覆盖bootstrap servers属性。

现在我想对自己实现的GenericContainer子类SchemaRegistryContainer也用@ServiceConnection注解管理，但运行时抛出ConnectionDetailsNotFoundException: No ConnectionDetails found for source '@ServiceConnection source for Bean 'schemaRegistryContainer''异常。

我想为这个@ServiceConnection配置自定义的ConnectionDetails（或其他替代方案），这样SchemaRegistryContainer启动时能修改所需的应用属性，具体是要把spring.kafka.consumer.properties.schema-registry-url替换成容器的URL。

能不能通过@ServiceDefinition实现？我不想退回到用静态字段和@DynamicPropertySource的方式。

测试配置 #

@TestConfiguration(proxyBeanMethods = false)
public class ContainerConfig {
    public static final DockerImageName DEFAULT_IMAGE_NAME = DockerImageName.parse("confluentinc/cp-kafka");
    public static final String CONFLUENT_PLATFORM_VERSION = "7.4.1";

    @Bean
    @ServiceConnection
    KafkaContainer kafkaContainer() {
        return new KafkaContainer(DEFAULT_IMAGE_NAME.withTag(CONFLUENT_PLATFORM_VERSION));
    }

    @Bean
    @ServiceConnection
    SchemaRegistryContainer schemaRegistryContainer(KafkaContainer kafkaContainer) {
        return new SchemaRegistryContainer(CONFLUENT_PLATFORM_VERSION, kafkaContainer);
    }
}

SchemaRegistryContainer 代码 #

public class SchemaRegistryContainer extends GenericContainer<SchemaRegistryContainer> {

    private static final int PORT = 8081;

    public SchemaRegistryContainer(String confluentVersion, KafkaContainer kafka) {
        super("confluentinc/cp-schema-registry:" + confluentVersion);

        withExposedPorts(PORT);
        dependsOn(kafka);
        withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", kafka.getBootstrapServers());
    }

    public String getSchemaRegistryUrl() {
        return "http://" + getHost() + ":" + getMappedPort(PORT);
    }
}

要让自定义的SchemaRegistryContainer支持@ServiceConnection，无需使用@ServiceDefinition，只需实现自定义ConnectionDetails和对应的工厂类即可，步骤如下：

1. 定义自定义ConnectionDetails #

创建实现ConnectionDetails的类，封装Schema Registry的连接信息，通过注解指定要覆盖的配置属性：

public class SchemaRegistryConnectionDetails implements ConnectionDetails {

    private final String schemaRegistryUrl;

    public SchemaRegistryConnectionDetails(String schemaRegistryUrl) {
        this.schemaRegistryUrl = schemaRegistryUrl;
    }

    @ConnectionDetailsKey("spring.kafka.consumer.properties.schema-registry-url")
    public String getSchemaRegistryUrl() {
        return schemaRegistryUrl;
    }
}

@ConnectionDetailsKey直接对应你需要替换的目标配置键，Spring Boot会自动用该值覆盖对应属性。

2. 实现ServiceConnectionDetailsFactory #

创建工厂类，将SchemaRegistryContainer实例转换为上面定义的ConnectionDetails：

@Component
public class SchemaRegistryConnectionDetailsFactory implements ServiceConnectionDetailsFactory<SchemaRegistryContainer> {

    @Override
    public boolean supports(Class<?> source) {
        return SchemaRegistryContainer.class.isAssignableFrom(source);
    }

    @Override
    public SchemaRegistryConnectionDetails getConnectionDetails(SchemaRegistryContainer source) {
        return new SchemaRegistryConnectionDetails(source.getSchemaRegistryUrl());
    }
}

Spring Boot会自动检测到这个工厂类，当发现带有@ServiceConnection的SchemaRegistryContainer Bean时，自动调用它生成连接信息，完成属性覆盖。

3. 保留原有配置 #

你的ContainerConfig无需修改，保持@ServiceConnection注解在SchemaRegistryContainer的Bean方法上即可。

这样配置后，Spring Boot会自动识别自定义容器，通过工厂生成的ConnectionDetails自动替换spring.kafka.consumer.properties.schema-registry-url属性，完全不用退回到@DynamicPropertySource或静态字段的方式。

内容的提问来源于stack exchange，提问作者matejs