Skip to content

Commit f0e3a0f

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 f0e3a0f

File tree

7 files changed

+251
-131
lines changed

7 files changed

+251
-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: 58 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,10 @@ 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).build().await.unwrap();
107116

108117
// do NOT create the topic
109118

@@ -167,8 +176,8 @@ async fn test_tls() {
167176
.with_single_cert(vec![producer_root], private_key)
168177
.unwrap();
169178

170-
let connection = maybe_skip_kafka_integration!();
171-
ClientBuilder::new(connection)
179+
let test_cfg = maybe_skip_kafka_integration!();
180+
ClientBuilder::new(test_cfg.bootstrap_brokers)
172181
.tls_config(Arc::new(config))
173182
.build()
174183
.await
@@ -180,14 +189,11 @@ async fn test_tls() {
180189
async fn test_socks5() {
181190
maybe_start_logging();
182191

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!();
192+
let test_cfg = maybe_skip_kafka_integration!(socks5);
187193
let topic_name = random_topic_name();
188194

189-
let client = ClientBuilder::new(connection)
190-
.socks5_proxy(proxy)
195+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
196+
.socks5_proxy(test_cfg.socks5_proxy.unwrap())
191197
.build()
192198
.await
193199
.unwrap();
@@ -222,11 +228,14 @@ async fn test_socks5() {
222228
async fn test_produce_empty() {
223229
maybe_start_logging();
224230

225-
let connection = maybe_skip_kafka_integration!();
231+
let test_cfg = maybe_skip_kafka_integration!();
226232
let topic_name = random_topic_name();
227233
let n_partitions = 2;
228234

229-
let client = ClientBuilder::new(connection).build().await.unwrap();
235+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
236+
.build()
237+
.await
238+
.unwrap();
230239
let controller_client = client.controller_client().unwrap();
231240
controller_client
232241
.create_topic(&topic_name, n_partitions, 1, 5_000)
@@ -247,11 +256,14 @@ async fn test_produce_empty() {
247256
async fn test_consume_empty() {
248257
maybe_start_logging();
249258

250-
let connection = maybe_skip_kafka_integration!();
259+
let test_cfg = maybe_skip_kafka_integration!();
251260
let topic_name = random_topic_name();
252261
let n_partitions = 2;
253262

254-
let client = ClientBuilder::new(connection).build().await.unwrap();
263+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
264+
.build()
265+
.await
266+
.unwrap();
255267
let controller_client = client.controller_client().unwrap();
256268
controller_client
257269
.create_topic(&topic_name, n_partitions, 1, 5_000)
@@ -274,11 +286,14 @@ async fn test_consume_empty() {
274286
async fn test_consume_offset_out_of_range() {
275287
maybe_start_logging();
276288

277-
let connection = maybe_skip_kafka_integration!();
289+
let test_cfg = maybe_skip_kafka_integration!();
278290
let topic_name = random_topic_name();
279291
let n_partitions = 2;
280292

281-
let client = ClientBuilder::new(connection).build().await.unwrap();
293+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
294+
.build()
295+
.await
296+
.unwrap();
282297
let controller_client = client.controller_client().unwrap();
283298
controller_client
284299
.create_topic(&topic_name, n_partitions, 1, 5_000)
@@ -314,11 +329,11 @@ async fn test_consume_offset_out_of_range() {
314329
async fn test_get_offset() {
315330
maybe_start_logging();
316331

317-
let connection = maybe_skip_kafka_integration!();
332+
let test_cfg = maybe_skip_kafka_integration!();
318333
let topic_name = random_topic_name();
319334
let n_partitions = 1;
320335

321-
let client = ClientBuilder::new(connection.clone())
336+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers.clone())
322337
.build()
323338
.await
324339
.unwrap();
@@ -382,10 +397,13 @@ async fn test_get_offset() {
382397
async fn test_produce_consume_size_cutoff() {
383398
maybe_start_logging();
384399

385-
let connection = maybe_skip_kafka_integration!();
400+
let test_cfg = maybe_skip_kafka_integration!();
386401
let topic_name = random_topic_name();
387402

388-
let client = ClientBuilder::new(connection).build().await.unwrap();
403+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
404+
.build()
405+
.await
406+
.unwrap();
389407
let controller_client = client.controller_client().unwrap();
390408
controller_client
391409
.create_topic(&topic_name, 1, 1, 5_000)
@@ -460,10 +478,13 @@ async fn test_produce_consume_size_cutoff() {
460478
async fn test_consume_midbatch() {
461479
maybe_start_logging();
462480

463-
let connection = maybe_skip_kafka_integration!();
481+
let test_cfg = maybe_skip_kafka_integration!();
464482
let topic_name = random_topic_name();
465483

466-
let client = ClientBuilder::new(connection).build().await.unwrap();
484+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
485+
.build()
486+
.await
487+
.unwrap();
467488
let controller_client = client.controller_client().unwrap();
468489
controller_client
469490
.create_topic(&topic_name, 1, 1, 5_000)
@@ -508,10 +529,13 @@ async fn test_consume_midbatch() {
508529
async fn test_delete_records() {
509530
maybe_start_logging();
510531

511-
let connection = maybe_skip_kafka_integration!();
532+
let test_cfg = maybe_skip_kafka_integration!(delete);
512533
let topic_name = random_topic_name();
513534

514-
let client = ClientBuilder::new(connection).build().await.unwrap();
535+
let client = ClientBuilder::new(test_cfg.bootstrap_brokers)
536+
.build()
537+
.await
538+
.unwrap();
515539
let controller_client = client.controller_client().unwrap();
516540
controller_client
517541
.create_topic(&topic_name, 1, 1, 5_000)
@@ -555,7 +579,10 @@ async fn test_delete_records() {
555579
let offset_4 = offsets[0];
556580

557581
// delete from the middle of the 2nd batch
558-
maybe_skip_delete!(partition_client, offset_3);
582+
partition_client
583+
.delete_records(offset_3, 1_000)
584+
.await
585+
.unwrap();
559586

560587
// fetching data before the record fails
561588
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)