0
# Testing Annotations
1

2
Annotation-based configuration for embedded Kafka in Spring tests, providing declarative setup with automatic broker lifecycle management and Spring TestContext integration.
3

4
## Capabilities
5

6
### EmbeddedKafka Annotation
7

8
Primary annotation for enabling embedded Kafka in Spring tests with comprehensive configuration options.
9

10
```java { .api }
11
/**
12
 * Annotation that can be specified on a test class that runs Spring for Apache Kafka based tests
13
 */
14
@ExtendWith(EmbeddedKafkaCondition.class)
15
@Target(ElementType.TYPE)
16
@Retention(RetentionPolicy.RUNTIME)
17
@Documented
18
@Inherited
19
@DisabledInAotMode
20
public @interface EmbeddedKafka {
21
    /**
22
     * The number of brokers
23
     * @return the number of brokers
24
     */
25
    @AliasFor("count")
26
    int value() default 1;
27

28
    /**
29
     * The number of brokers
30
     * @return the number of brokers
31
     */
32
    @AliasFor("value")
33
    int count() default 1;
34

35
    /**
36
     * Passed into kafka.utils.TestUtils.createBrokerConfig()
37
     * @return controlled shutdown flag
38
     */
39
    boolean controlledShutdown() default false;
40

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

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

56
    /**
57
     * Partitions per topic
58
     * @return partitions per topic
59
     */
60
    int partitions() default 2;
61

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

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

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

87
    /**
88
     * The property name to set with the bootstrap server addresses
89
     * Defaults to spring.kafka.bootstrap-servers
90
     * @return the property name
91
     */
92
    String bootstrapServersProperty() default "spring.kafka.bootstrap-servers";
93

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

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

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

114
    /**
115
     * Use KRaft instead of Zookeeper; default false
116
     * @return whether to use KRaft
117
     */
118
    boolean kraft() default false;
119
}
120
```
121

122
### Context Customization
123

124
Spring TestContext integration components for automatic embedded Kafka setup.
125

126
```java { .api }
127
/**
128
 * Spring test context customizer for embedded Kafka
129
 */
130
public class EmbeddedKafkaContextCustomizer implements ContextCustomizer {
131
    // Implementation details handled automatically by Spring TestContext
132
}
133

134
/**
135
 * Factory for creating embedded Kafka context customizers
136
 */
137
public class EmbeddedKafkaContextCustomizerFactory implements ContextCustomizerFactory {
138
    /**
139
     * Create context customizer if EmbeddedKafka annotation is present
140
     * @param testClass the test class
141
     * @param configAttributesList the configuration attributes
142
     * @return context customizer or null
143
     */
144
    public ContextCustomizer createContextCustomizer(Class<?> testClass, List<ContextConfigurationAttributes> configAttributesList);
145
}
146
```
147

148
**Usage Examples:**
149

150
```java
151
// Basic embedded Kafka setup
152
@SpringBootTest
153
@EmbeddedKafka(partitions = 1, topics = { "test-topic" })
154
public class BasicKafkaTest {
155
    @Autowired
156
    private EmbeddedKafkaBroker embeddedKafka;
157
    
158
    @Test
159
    public void testBasicKafka() {
160
        // Test implementation
161
    }
162
}
163

164
// Multiple brokers with custom configuration
165
@SpringBootTest
166
@EmbeddedKafka(
167
    count = 3, 
168
    partitions = 2,
169
    topics = { "orders", "payments", "notifications" },
170
    brokerProperties = {
171
        "auto.create.topics.enable=true",
172
        "transaction.state.log.replication.factor=1"
173
    }
174
)
175
public class MultiBrokerKafkaTest {
176
    @Autowired
177
    private EmbeddedKafkaBroker embeddedKafka;
178
    
179
    @Test
180
    public void testMultiBrokerSetup() {
181
        assertEquals(3, embeddedKafka.getKafkaServers().size());
182
        assertTrue(embeddedKafka.getTopics().contains("orders"));
183
    }
184
}
185

186
// KRaft mode (ZooKeeper-free)
187
@SpringBootTest
188
@EmbeddedKafka(
189
    kraft = true,
190
    partitions = 1,
191
    topics = { "kraft-topic" },
192
    brokerProperties = { "num.partitions=1" }
193
)
194
public class KRaftKafkaTest {
195
    @Autowired
196
    private EmbeddedKafkaBroker embeddedKafka;
197
    
198
    @Test
199
    public void testKRaftMode() {
200
        // KRaft-specific test implementation
201
    }
202
}
203

204
// Custom ports (ZooKeeper mode only)
205
@SpringBootTest
206
@EmbeddedKafka(
207
    ports = { 9092, 9093 },
208
    zookeeperPort = 2181,
209
    count = 2,
210
    topics = { "fixed-port-topic" }
211
)
212
public class FixedPortKafkaTest {
213
    @Autowired
214
    private EmbeddedKafkaBroker embeddedKafka;
215
    
216
    @Test
217
    public void testFixedPorts() {
218
        assertTrue(embeddedKafka.getBrokersAsString().contains("9092"));
219
        assertTrue(embeddedKafka.getBrokersAsString().contains("9093"));
220
    }
221
}
222

223
// Property placeholders with Spring context
224
@SpringBootTest
225
@EmbeddedKafka(
226
    topics = { "${kafka.topic.orders:default-orders}", "${kafka.topic.events:default-events}" },
227
    brokerProperties = { "delete.topic.enable=${kafka.delete.enable:true}" },
228
    bootstrapServersProperty = "${kafka.bootstrap.property:spring.kafka.bootstrap-servers}"
229
)
230
@TestPropertySource(properties = {
231
    "kafka.topic.orders=test-orders",
232
    "kafka.topic.events=test-events",
233
    "kafka.delete.enable=false"
234
})
235
public class PropertyPlaceholderKafkaTest {
236
    @Autowired
237
    private EmbeddedKafkaBroker embeddedKafka;
238
    
239
    @Test
240
    public void testPropertyPlaceholders() {
241
        assertTrue(embeddedKafka.getTopics().contains("test-orders"));
242
        assertTrue(embeddedKafka.getTopics().contains("test-events"));
243
    }
244
}
245

246
// External properties file
247
@SpringBootTest
248
@EmbeddedKafka(
249
    topics = { "config-topic" },
250
    brokerPropertiesLocation = "classpath:kafka-test.properties"
251
)
252
public class ExternalPropertiesKafkaTest {
253
    // kafka-test.properties contains:
254
    // auto.create.topics.enable=true
255
    // transaction.state.log.replication.factor=1
256
    // offsets.topic.replication.factor=1
257
}
258
```