Skip to content

Commit 6565321

Browse files
crepererumcrepererum
committed
test: improve testing config
Instead of having X different macros and control flow exists, let's have one mechanism to read env variables and decide if a test shall run or not. This also moves the "which broker implementation supports which features" from a bunch of environment variables into a nice enum. This will greatly help when we expand the set of admit APIs, see #157 for example.
1 parent 2f855b2 commit 6565321

File tree

7 files changed

+254
-131
lines changed

7 files changed

+254
-131
lines changed

‎.circleci/config.yml

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -207,6 +207,7 @@ jobs:
207207
RUST_LOG: "trace"
208208
# Run integration tests
209209
TEST_INTEGRATION: 1
210+
TEST_BROKER_IMPL: redpanda
210211
TEST_JAVA_INTEROPT: 1
211212
# Don't use the first node here since this is likely the controller and we want to ensure that we automatically
212213
# pick the controller for certain actions (e.g. topic creation) and don't just get lucky.
@@ -284,8 +285,7 @@ jobs:
284285
RUST_LOG: "trace"
285286
# Run integration tests
286287
TEST_INTEGRATION: 1
287-
# Kafka support DeleteRecords
288-
TEST_DELETE_RECORDS: 1
288+
TEST_BROKER_IMPL: kafka
289289
TEST_JAVA_INTEROPT: 1
290290
# Don't use the first node here since this is likely the controller and we want to ensure that we automatically
291291
# pick the controller for certain actions (e.g. topic creation) and don't just get lucky.

‎README.md

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@ $ docker-compose -f docker-compose-redpanda.yml up
115115
in one session, and then run:
116116

117117
```console
118-
$ TEST_INTEGRATION=1 KAFKA_CONNECT=0.0.0.0:9011 cargo test
118+
$ TEST_INTEGRATION=1 TEST_BROKER_IMPL=redpanda KAFKA_CONNECT=0.0.0.0:9011 cargo test
119119
```
120120

121121
in another session.
@@ -131,7 +131,7 @@ $ docker-compose -f docker-compose-kafka.yml up
131131
in one session, and then run:
132132

133133
```console
134-
$ TEST_INTEGRATION=1 TEST_DELETE_RECORDS=1 KAFKA_CONNECT=localhost:9011 cargo test
134+
$ TEST_INTEGRATION=1 TEST_BROKER_IMPL=kafka KAFKA_CONNECT=localhost:9011 cargo test
135135
```
136136

137137
in another session. Note that Apache Kafka supports a different set of features then redpanda, so we pass other
@@ -231,14 +231,14 @@ execution that hooks right into the place where it is about to exit:
231231
Install [cargo-criterion], make sure you have some Kafka cluster running, and then you can run all benchmarks with:
232232

233233
```console
234-
$ TEST_INTEGRATION=1 KAFKA_CONNECT=localhost:9011 cargo criterion --all-features
234+
$ TEST_INTEGRATION=1 TEST_BROKER_IMPL=kafka KAFKA_CONNECT=localhost:9011 cargo criterion --all-features
235235
```
236236

237237
If you find a benchmark that is too slow, you can may want to profile it. Get [cargo-with], and [perf], then run (here
238238
for the `parallel/rskafka` benchmark):
239239

240240
```console
241-
$ TEST_INTEGRATION=1 KAFKA_CONNECT=localhost:9011 cargo with 'perf record --call-graph dwarf -- {bin}' -- \
241+
$ TEST_INTEGRATION=1 TEST_BROKER_IMPL=kafka KAFKA_CONNECT=localhost:9011 cargo with 'perf record --call-graph dwarf -- {bin}' -- \
242242
bench --all-features --bench write_throughput -- \
243243
--bench --noplot parallel/rskafka
244244
```

‎tests/client.rs

Lines changed: 61 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -16,16 +16,22 @@ use test_helpers::{maybe_start_logging, now, random_topic_name, record};
1616
async fn test_plain() {
1717
maybe_start_logging();
1818

19-
let connection = maybe_skip_kafka_integration!();
20-
ClientBuilder::new(connection).build().await.unwrap();
19+
let test_cfg = maybe_skip_kafka_integration!();
20+
ClientBuilder::new(test_cfg.bootstrap_brokers)
21+
.build()
22+
.await
23+
.unwrap();
2124
}
2225

2326
#[tokio::test]
2427
async fn test_topic_crud() {
2528
maybe_start_logging();
2629

27-
let connection = maybe_skip_kafka_integration!();
28-
let client = ClientBuilder::new(connection).build().await.unwrap();
30+
let test_cfg = maybe_skip_kafka_integration!();
31+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
32+
.build()
33+
.await
34+
.unwrap();
2935
let controller_client = client.controller_client().unwrap();
3036
let topics = client.list_topics().await.unwrap();
3137

@@ -77,10 +83,13 @@ async fn test_topic_crud() {
7783
async fn test_partition_client() {
7884
maybe_start_logging();
7985

80-
let connection = maybe_skip_kafka_integration!();
86+
let test_cfg = maybe_skip_kafka_integration!();
8187
let topic_name = random_topic_name();
8288

83-
let client = ClientBuilder::new(connection).build().await.unwrap();
89+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
90+
.build()
91+
.await
92+
.unwrap();
8493

8594
let controller_client = client.controller_client().unwrap();
8695
controller_client
@@ -100,10 +109,13 @@ async fn test_partition_client() {
100109
async fn test_non_existing_partition() {
101110
maybe_start_logging();
102111

103-
let connection = maybe_skip_kafka_integration!();
112+
let test_cfg = maybe_skip_kafka_integration!();
104113
let topic_name = random_topic_name();
105114

106-
let client = ClientBuilder::new(connection).build().await.unwrap();
115+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
116+
.build()
117+
.await
118+
.unwrap();
107119

108120
// do NOT create the topic
109121

@@ -167,8 +179,8 @@ async fn test_tls() {
167179
.with_single_cert(vec![producer_root], private_key)
168180
.unwrap();
169181

170-
let connection = maybe_skip_kafka_integration!();
171-
ClientBuilder::new(connection)
182+
let test_cfg = maybe_skip_kafka_integration!();
183+
ClientBuilder::new(test_cfg.bootstrap_brokers)
172184
.tls_config(Arc::new(config))
173185
.build()
174186
.await
@@ -180,14 +192,11 @@ async fn test_tls() {
180192
async fn test_socks5() {
181193
maybe_start_logging();
182194

183-
// e.g. "my-connection-kafka-bootstrap:9092"
184-
let connection = maybe_skip_kafka_integration!();
185-
// e.g. "localhost:1080"
186-
let proxy = maybe_skip_SOCKS_PROXY!();
195+
let test_cfg = maybe_skip_kafka_integration!(socks5);
187196
let topic_name = random_topic_name();
188197

189-
let client = ClientBuilder::new(connection)
190-
.socks5_proxy(proxy)
198+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
199+
.socks5_proxy(test_cfg.socks5_proxy.unwrap())
191200
.build()
192201
.await
193202
.unwrap();
@@ -222,11 +231,14 @@ async fn test_socks5() {
222231
async fn test_produce_empty() {
223232
maybe_start_logging();
224233

225-
let connection = maybe_skip_kafka_integration!();
234+
let test_cfg = maybe_skip_kafka_integration!();
226235
let topic_name = random_topic_name();
227236
let n_partitions = 2;
228237

229-
let client = ClientBuilder::new(connection).build().await.unwrap();
238+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
239+
.build()
240+
.await
241+
.unwrap();
230242
let controller_client = client.controller_client().unwrap();
231243
controller_client
232244
.create_topic(&topic_name, n_partitions, 1, 5_000)
@@ -247,11 +259,14 @@ async fn test_produce_empty() {
247259
async fn test_consume_empty() {
248260
maybe_start_logging();
249261

250-
let connection = maybe_skip_kafka_integration!();
262+
let test_cfg = maybe_skip_kafka_integration!();
251263
let topic_name = random_topic_name();
252264
let n_partitions = 2;
253265

254-
let client = ClientBuilder::new(connection).build().await.unwrap();
266+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
267+
.build()
268+
.await
269+
.unwrap();
255270
let controller_client = client.controller_client().unwrap();
256271
controller_client
257272
.create_topic(&topic_name, n_partitions, 1, 5_000)
@@ -274,11 +289,14 @@ async fn test_consume_empty() {
274289
async fn test_consume_offset_out_of_range() {
275290
maybe_start_logging();
276291

277-
let connection = maybe_skip_kafka_integration!();
292+
let test_cfg = maybe_skip_kafka_integration!();
278293
let topic_name = random_topic_name();
279294
let n_partitions = 2;
280295

281-
let client = ClientBuilder::new(connection).build().await.unwrap();
296+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
297+
.build()
298+
.await
299+
.unwrap();
282300
let controller_client = client.controller_client().unwrap();
283301
controller_client
284302
.create_topic(&topic_name, n_partitions, 1, 5_000)
@@ -314,11 +332,11 @@ async fn test_consume_offset_out_of_range() {
314332
async fn test_get_offset() {
315333
maybe_start_logging();
316334

317-
let connection = maybe_skip_kafka_integration!();
335+
let test_cfg = maybe_skip_kafka_integration!();
318336
let topic_name = random_topic_name();
319337
let n_partitions = 1;
320338

321-
let client = ClientBuilder::new(connection.clone())
339+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers.clone())
322340
.build()
323341
.await
324342
.unwrap();
@@ -382,10 +400,13 @@ async fn test_get_offset() {
382400
async fn test_produce_consume_size_cutoff() {
383401
maybe_start_logging();
384402

385-
let connection = maybe_skip_kafka_integration!();
403+
let test_cfg = maybe_skip_kafka_integration!();
386404
let topic_name = random_topic_name();
387405

388-
let client = ClientBuilder::new(connection).build().await.unwrap();
406+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
407+
.build()
408+
.await
409+
.unwrap();
389410
let controller_client = client.controller_client().unwrap();
390411
controller_client
391412
.create_topic(&topic_name, 1, 1, 5_000)
@@ -460,10 +481,13 @@ async fn test_produce_consume_size_cutoff() {
460481
async fn test_consume_midbatch() {
461482
maybe_start_logging();
462483

463-
let connection = maybe_skip_kafka_integration!();
484+
let test_cfg = maybe_skip_kafka_integration!();
464485
let topic_name = random_topic_name();
465486

466-
let client = ClientBuilder::new(connection).build().await.unwrap();
487+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
488+
.build()
489+
.await
490+
.unwrap();
467491
let controller_client = client.controller_client().unwrap();
468492
controller_client
469493
.create_topic(&topic_name, 1, 1, 5_000)
@@ -508,10 +532,13 @@ async fn test_consume_midbatch() {
508532
async fn test_delete_records() {
509533
maybe_start_logging();
510534

511-
let connection = maybe_skip_kafka_integration!();
535+
let test_cfg = maybe_skip_kafka_integration!(delete);
512536
let topic_name = random_topic_name();
513537

514-
let client = ClientBuilder::new(connection).build().await.unwrap();
538+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
539+
.build()
540+
.await
541+
.unwrap();
515542
let controller_client = client.controller_client().unwrap();
516543
controller_client
517544
.create_topic(&topic_name, 1, 1, 5_000)
@@ -555,7 +582,10 @@ async fn test_delete_records() {
555582
let offset_4 = offsets[0];
556583

557584
// delete from the middle of the 2nd batch
558-
maybe_skip_delete!(partition_client, offset_3);
585+
partition_client
586+
.delete_records(offset_3, 1_000)
587+
.await
588+
.unwrap();
559589

560590
// fetching data before the record fails
561591
let err = partition_client

‎tests/consumer.rs

Lines changed: 46 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,11 @@ mod test_helpers;
2222
async fn test_stream_consumer_start_at_0() {
2323
maybe_start_logging();
2424

25-
let connection = maybe_skip_kafka_integration!();
26-
let client = ClientBuilder::new(connection).build().await.unwrap();
25+
let test_cfg = maybe_skip_kafka_integration!();
26+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
27+
.build()
28+
.await
29+
.unwrap();
2730
let controller_client = client.controller_client().unwrap();
2831

2932
let topic = random_topic_name();
@@ -77,8 +80,11 @@ async fn test_stream_consumer_start_at_0() {
7780
async fn test_stream_consumer_start_at_1() {
7881
maybe_start_logging();
7982

80-
let connection = maybe_skip_kafka_integration!();
81-
let client = ClientBuilder::new(connection).build().await.unwrap();
83+
let test_cfg = maybe_skip_kafka_integration!();
84+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
85+
.build()
86+
.await
87+
.unwrap();
8288
let controller_client = client.controller_client().unwrap();
8389

8490
let topic = random_topic_name();
@@ -121,8 +127,11 @@ async fn test_stream_consumer_start_at_1() {
121127
async fn test_stream_consumer_offset_out_of_range() {
122128
maybe_start_logging();
123129

124-
let connection = maybe_skip_kafka_integration!();
125-
let client = ClientBuilder::new(connection).build().await.unwrap();
130+
let test_cfg = maybe_skip_kafka_integration!();
131+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
132+
.build()
133+
.await
134+
.unwrap();
126135
let controller_client = client.controller_client().unwrap();
127136

128137
let topic = random_topic_name();
@@ -157,8 +166,11 @@ async fn test_stream_consumer_offset_out_of_range() {
157166
async fn test_stream_consumer_start_at_earliest() {
158167
maybe_start_logging();
159168

160-
let connection = maybe_skip_kafka_integration!();
161-
let client = ClientBuilder::new(connection).build().await.unwrap();
169+
let test_cfg = maybe_skip_kafka_integration!();
170+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
171+
.build()
172+
.await
173+
.unwrap();
162174
let controller_client = client.controller_client().unwrap();
163175

164176
let topic = random_topic_name();
@@ -212,8 +224,11 @@ async fn test_stream_consumer_start_at_earliest() {
212224
async fn test_stream_consumer_start_at_earliest_empty() {
213225
maybe_start_logging();
214226

215-
let connection = maybe_skip_kafka_integration!();
216-
let client = ClientBuilder::new(connection).build().await.unwrap();
227+
let test_cfg = maybe_skip_kafka_integration!();
228+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
229+
.build()
230+
.await
231+
.unwrap();
217232
let controller_client = client.controller_client().unwrap();
218233

219234
let topic = random_topic_name();
@@ -257,8 +272,16 @@ async fn test_stream_consumer_start_at_earliest_empty() {
257272
async fn test_stream_consumer_start_at_earliest_after_deletion() {
258273
maybe_start_logging();
259274

260-
let connection = maybe_skip_kafka_integration!();
261-
let client = ClientBuilder::new(connection).build().await.unwrap();
275+
let test_cfg = maybe_skip_kafka_integration!(delete);
276+
if !test_cfg.broker_impl.supports_deletes() {
277+
println!("Skipping due to missing delete support");
278+
return;
279+
}
280+
281+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
282+
.build()
283+
.await
284+
.unwrap();
262285
let controller_client = client.controller_client().unwrap();
263286

264287
let topic = random_topic_name();
@@ -284,7 +307,7 @@ async fn test_stream_consumer_start_at_earliest_after_deletion() {
284307
.await
285308
.unwrap();
286309

287-
maybe_skip_delete!(partition_client, 1);
310+
partition_client.delete_records(1, 1_000).await.unwrap();
288311

289312
let mut stream =
290313
StreamConsumerBuilder::new(Arc::clone(&partition_client), StartOffset::Earliest)
@@ -304,8 +327,11 @@ async fn test_stream_consumer_start_at_earliest_after_deletion() {
304327
async fn test_stream_consumer_start_at_latest() {
305328
maybe_start_logging();
306329

307-
let connection = maybe_skip_kafka_integration!();
308-
let client = ClientBuilder::new(connection).build().await.unwrap();
330+
let test_cfg = maybe_skip_kafka_integration!();
331+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
332+
.build()
333+
.await
334+
.unwrap();
309335
let controller_client = client.controller_client().unwrap();
310336

311337
let topic = random_topic_name();
@@ -353,8 +379,11 @@ async fn test_stream_consumer_start_at_latest() {
353379
async fn test_stream_consumer_start_at_latest_empty() {
354380
maybe_start_logging();
355381

356-
let connection = maybe_skip_kafka_integration!();
357-
let client = ClientBuilder::new(connection).build().await.unwrap();
382+
let test_cfg = maybe_skip_kafka_integration!();
383+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
384+
.build()
385+
.await
386+
.unwrap();
358387
let controller_client = client.controller_client().unwrap();
359388

360389
let topic = random_topic_name();

0 commit comments

Comments
 (0)