Testing Annotations

/**
 * Annotation that can be specified on a test class that runs Spring for Apache Kafka based tests
 */
@ExtendWith(EmbeddedKafkaCondition.class)
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@DisabledInAotMode
public @interface EmbeddedKafka {
    /**
     * The number of brokers
     * @return the number of brokers
     */
    @AliasFor("count")
    int value() default 1;

    /**
     * The number of brokers
     * @return the number of brokers
     */
    @AliasFor("value")
    int count() default 1;

    /**
     * Passed into kafka.utils.TestUtils.createBrokerConfig()
     * @return controlled shutdown flag
     */
    boolean controlledShutdown() default false;

    /**
     * Set explicit ports on which the kafka brokers will listen
     * A port must be provided for each instance, which means the number of ports must match the value of the count attribute
     * This property is not valid when using KRaft mode
     * @return ports for brokers
     */
    int[] ports() default { 0 };

    /**
     * Set the port on which the embedded Zookeeper should listen
     * This property is not valid when using KRaft mode
     * @return the port
     */
    int zookeeperPort() default 0;

    /**
     * Partitions per topic
     * @return partitions per topic
     */
    int partitions() default 2;

    /**
     * Topics that should be created. Topics may contain property place holders
     * The topics will be created with partitions() partitions
     * Place holders will only be resolved when there is a Spring test application context present
     * @return the topics to create
     */
    String[] topics() default { };

    /**
     * Properties in form key=value that should be added to the broker config before runs
     * When used in a Spring test context, properties may contain property place holders
     * Place holders will only be resolved when there is a Spring test application context present
     * @return the properties to add
     */
    String[] brokerProperties() default { };

    /**
     * Spring Resource url specifying the location of properties that should be added to the broker config
     * When used in a Spring test context, the brokerPropertiesLocation url and the properties themselves may contain place holders that are resolved during initialization
     * Properties specified by brokerProperties() will override properties found in brokerPropertiesLocation
     * Place holders will only be resolved when there is a Spring test application context present
     * @return a Resource url specifying the location of properties to add
     */
    String brokerPropertiesLocation() default "";

    /**
     * The property name to set with the bootstrap server addresses
     * Defaults to spring.kafka.bootstrap-servers
     * @return the property name
     */
    String bootstrapServersProperty() default "spring.kafka.bootstrap-servers";

    /**
     * Timeout for internal ZK client connection
     * This property is not valid when using KRaft mode
     * @return default DEFAULT_ZK_CONNECTION_TIMEOUT
     */
    int zkConnectionTimeout() default EmbeddedKafkaZKBroker.DEFAULT_ZK_CONNECTION_TIMEOUT;

    /**
     * Timeout for internal ZK client session
     * This property is not valid when using KRaft mode
     * @return default DEFAULT_ZK_SESSION_TIMEOUT
     */
    int zkSessionTimeout() default EmbeddedKafkaZKBroker.DEFAULT_ZK_SESSION_TIMEOUT;

    /**
     * Timeout in seconds for admin operations (e.g. topic creation, close)
     * @return default DEFAULT_ADMIN_TIMEOUT
     */
    int adminTimeout() default EmbeddedKafkaBroker.DEFAULT_ADMIN_TIMEOUT;

    /**
     * Use KRaft instead of Zookeeper; default false
     * @return whether to use KRaft
     */
    boolean kraft() default false;
}

/**
 * Spring test context customizer for embedded Kafka
 */
public class EmbeddedKafkaContextCustomizer implements ContextCustomizer {
    // Implementation details handled automatically by Spring TestContext
}

/**
 * Factory for creating embedded Kafka context customizers
 */
public class EmbeddedKafkaContextCustomizerFactory implements ContextCustomizerFactory {
    /**
     * Create context customizer if EmbeddedKafka annotation is present
     * @param testClass the test class
     * @param configAttributesList the configuration attributes
     * @return context customizer or null
     */
    public ContextCustomizer createContextCustomizer(Class<?> testClass, List<ContextConfigurationAttributes> configAttributesList);
}

// Basic embedded Kafka setup
@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = { "test-topic" })
public class BasicKafkaTest {
    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;
    
    @Test
    public void testBasicKafka() {
        // Test implementation
    }
}

// Multiple brokers with custom configuration
@SpringBootTest
@EmbeddedKafka(
    count = 3, 
    partitions = 2,
    topics = { "orders", "payments", "notifications" },
    brokerProperties = {
        "auto.create.topics.enable=true",
        "transaction.state.log.replication.factor=1"
    }
)
public class MultiBrokerKafkaTest {
    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;
    
    @Test
    public void testMultiBrokerSetup() {
        assertEquals(3, embeddedKafka.getKafkaServers().size());
        assertTrue(embeddedKafka.getTopics().contains("orders"));
    }
}

// KRaft mode (ZooKeeper-free)
@SpringBootTest
@EmbeddedKafka(
    kraft = true,
    partitions = 1,
    topics = { "kraft-topic" },
    brokerProperties = { "num.partitions=1" }
)
public class KRaftKafkaTest {
    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;
    
    @Test
    public void testKRaftMode() {
        // KRaft-specific test implementation
    }
}

// Custom ports (ZooKeeper mode only)
@SpringBootTest
@EmbeddedKafka(
    ports = { 9092, 9093 },
    zookeeperPort = 2181,
    count = 2,
    topics = { "fixed-port-topic" }
)
public class FixedPortKafkaTest {
    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;
    
    @Test
    public void testFixedPorts() {
        assertTrue(embeddedKafka.getBrokersAsString().contains("9092"));
        assertTrue(embeddedKafka.getBrokersAsString().contains("9093"));
    }
}

// Property placeholders with Spring context
@SpringBootTest
@EmbeddedKafka(
    topics = { "${kafka.topic.orders:default-orders}", "${kafka.topic.events:default-events}" },
    brokerProperties = { "delete.topic.enable=${kafka.delete.enable:true}" },
    bootstrapServersProperty = "${kafka.bootstrap.property:spring.kafka.bootstrap-servers}"
)
@TestPropertySource(properties = {
    "kafka.topic.orders=test-orders",
    "kafka.topic.events=test-events",
    "kafka.delete.enable=false"
})
public class PropertyPlaceholderKafkaTest {
    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;
    
    @Test
    public void testPropertyPlaceholders() {
        assertTrue(embeddedKafka.getTopics().contains("test-orders"));
        assertTrue(embeddedKafka.getTopics().contains("test-events"));
    }
}

// External properties file
@SpringBootTest
@EmbeddedKafka(
    topics = { "config-topic" },
    brokerPropertiesLocation = "classpath:kafka-test.properties"
)
public class ExternalPropertiesKafkaTest {
    // kafka-test.properties contains:
    // auto.create.topics.enable=true
    // transaction.state.log.replication.factor=1
    // offsets.topic.replication.factor=1
}