为JUnit 4/5创建注解初始化注入Kafkaesque测试对象是否可行？

I'm developing a Kafka testing library called Kafkaesque, which supports writing Kafka integration tests with a fluent, elegant API. Currently, I'm working on the Spring Kafka adaptation version. Right now, every test requires manual initialization of the SpringKafkaesque object:

@Test void consumeShouldConsumeMessagesProducesFromOutsideProducer() {
 kafkaTemplate.sendDefault(1, "data1");
 kafkaTemplate.sendDefault(2, "data2");
 new SpringKafkaesque(broker)
 .<Integer, String>consume()
 .fromTopic(CONSUMER_TEST_TOPIC)
 .waitingAtMost(1L, TimeUnit.SECONDS)
 .waitingEmptyPolls(5, 100L, TimeUnit.MILLISECONDS)
 .withDeserializers(new IntegerDeserializer(), new StringDeserializer())
 .expecting()
 .havingRecordsSize(2)
 .assertingThatPayloads(Matchers.containsInAnyOrder("data1", "data2"))
 .andCloseConsumer();
}

I want to create a @Kafkaesque annotation similar to Spring Kafka's @EmbeddedKafka, replacing manual initialization and enabling automatic injection of the Kafkaesque object. Here's the desired example:

@SpringBootTest(classes = {TestConfiguration.class})
@Kafkaesque( topics = {SpringKafkaesqueTest.CONSUMER_TEST_TOPIC, SpringKafkaesqueTest.PRODUCER_TEST_TOPIC})
class SpringKafkaesqueTest {
 @Autowired private Kafkaesque kafkaesque;

 @Test void consumeShouldConsumeMessagesProducesFromOutsideProducer() {
 kafkaTemplate.sendDefault(1, "data1");
 kafkaTemplate.sendDefault(2, "data2");
 kafkaesque
 .<Integer, String>consume()
 .fromTopic(CONSUMER_TEST_TOPIC)
 .waitingAtMost(1L, TimeUnit.SECONDS)
 .waitingEmptyPolls(5, 100L, TimeUnit.MILLISECONDS)
 .withDeserializers(new IntegerDeserializer(), new StringDeserializer())
 .expecting()
 .havingRecordsSize(2)
 .assertingThatPayloads(Matchers.containsInAnyOrder("data1", "data2"))
 .andCloseConsumer();
 }
}

Is this approach feasible? What implementation suggestions do you have?

Absolutely, this approach is not only feasible but also aligns perfectly with Spring's annotation-driven testing ecosystem—great idea to mirror @EmbeddedKafka's user experience! Here are concrete implementation suggestions to make this work:

1. Build the Custom @Kafkaesque Annotation #

First, define the annotation itself with attributes like topics (to auto-create test topics), optional broker configuration, and default serializer/deserializer classes to reduce boilerplate:

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import(KafkaesqueConfiguration.class)
public @interface Kafkaesque {
    String[] topics() default {};
    // Add optional attributes like brokerPort, autoCreateTopics, defaultKeyDeserializer, etc.
}

2. Create a Configuration Class to Wire Up Beans #

Implement KafkaesqueConfiguration (referenced in the annotation's @Import) to handle the core logic:

• Reuse Spring's Embedded Kafka: Leverage EmbeddedKafkaBroker from spring-kafka-test to auto-spin up an embedded cluster (you can also add support for external brokers later if needed).
• Auto-Create Topics: Use the topics attribute from @Kafkaesque to create specified topics on the embedded broker before tests run.
• Instantiate & Expose Kafkaesque Bean: Initialize SpringKafkaesque with the embedded broker and any default settings from the annotation, making it available for @Autowired injection.

Example snippet:

@Configuration
public class KafkaesqueConfiguration implements ApplicationContextAware {
    private ApplicationContext applicationContext;

    @Bean
    public Kafkaesque kafkaesque(KafkaesqueAnnotationProperties properties, EmbeddedKafkaBroker broker) {
        // Auto-create topics from the annotation
        Arrays.stream(properties.getTopics()).forEach(broker::addTopics);
        // Initialize SpringKafkaesque with pre-configured settings
        return new SpringKafkaesque(broker)
                .withDefaultDeserializers(properties.getDefaultKeyDeserializer(), properties.getDefaultValueDeserializer());
    }

    // Helper bean to extract properties from @Kafkaesque
    @Bean
    public KafkaesqueAnnotationProperties kafkaesqueAnnotationProperties() {
        Kafkaesque annotation = applicationContext.findAnnotationOnBean(Kafkaesque.class);
        if (annotation == null) {
            return new KafkaesqueAnnotationProperties(); // Fallback to defaults
        }
        return new KafkaesqueAnnotationProperties(
                annotation.topics(),
                annotation.defaultKeyDeserializer(),
                annotation.defaultValueDeserializer()
        );
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

3. Hook Into Spring Test's Lifecycle #

Ensure proper cleanup and state management:

• Use Spring's built-in context lifecycle to shut down the embedded broker automatically when tests finish.
• If you need to reset topic data between tests, implement a TestExecutionListener that hooks into beforeTestMethod events to clear topics or reset the Kafkaesque instance.

4. Add Flexible Customization Options #

Make the annotation adaptable to different test scenarios:

• Allow users to override the embedded broker's configuration (e.g., port, partitions) via annotation attributes.
• Support default serializers/deserializers so users don't have to specify them in every test chain.
• Add polling-related defaults (like default waitingAtMost duration) to reduce repetitive test code.

5. Validate Annotation Usage #

Add runtime checks to ensure the annotation is used correctly:

• Use a BeanPostProcessor or ApplicationListener to verify @Kafkaesque is only applied to test classes alongside Spring test annotations like @SpringBootTest.
• Throw meaningful exceptions if required attributes are missing or invalid.

6. Test the Annotation Internals #

Write integration tests for your @Kafkaesque annotation to verify:

• The Kafkaesque bean is correctly injected into test classes.
• Specified topics are auto-created on the embedded broker.
• The fluent API works as expected with the injected bean.
• Cleanup happens properly between test runs.

This setup will give users a seamless, intuitive experience just like @EmbeddedKafka, but tailored to your Kafkaesque library's unique capabilities.

内容的提问来源于stack exchange，提问作者riccardo.cardin