@@ -30,24 +30,24 @@ protected function setUp(): void
30
30
public function testCreateWithProducingAndConsuming (): void
31
31
{
32
32
$ clusterConfig = new Conf ();
33
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
33
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
34
34
$ cluster = MockCluster::create (1 , $ clusterConfig );
35
35
36
36
$ producerConfig = new Conf ();
37
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
37
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
38
38
$ producerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
39
39
$ producer = new Producer ($ producerConfig );
40
40
$ producerTopic = $ producer ->newTopic (KAFKA_TEST_TOPIC );
41
41
$ producerTopic ->produce (0 , 0 , __METHOD__ , __METHOD__ );
42
42
$ producer ->flush (KAFKA_TEST_LONG_TIMEOUT_MS );
43
43
44
44
$ consumerConfig = new Conf ();
45
- $ consumerConfig ->set ('log_level ' , (string ) LOG_EMERG );
45
+ $ consumerConfig ->set ('log_level ' , (string )LOG_EMERG );
46
46
$ consumerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
47
47
$ consumerConfig ->set ('group.id ' , __METHOD__ );
48
48
$ consumer = new KafkaConsumer ($ consumerConfig );
49
49
$ consumer ->assign ([new TopicPartition (KAFKA_TEST_TOPIC , 0 )]);
50
- $ message = $ consumer ->consume (KAFKA_TEST_SHORT_TIMEOUT_MS );
50
+ $ message = $ consumer ->consume (KAFKA_TEST_LONG_TIMEOUT_MS );
51
51
52
52
$ this ->assertSame (__METHOD__ , $ message ->payload );
53
53
$ this ->assertSame (__METHOD__ , $ message ->key );
@@ -65,8 +65,8 @@ public function testFromProducer(): void
65
65
$ this ->requiresLibrdkafkaVersion ('>= ' , '1.4.0 ' );
66
66
67
67
$ producerConfig = new Conf ();
68
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
69
- $ producerConfig ->set ('test.mock.num.brokers ' , (string ) 3 );
68
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
69
+ $ producerConfig ->set ('test.mock.num.brokers ' , (string )3 );
70
70
$ producer = new Producer ($ producerConfig );
71
71
72
72
$ cluster = MockCluster::fromProducer ($ producer );
@@ -79,7 +79,7 @@ public function testFromProducerWithMissingConfigShouldFail(): void
79
79
$ this ->requiresLibrdkafkaVersion ('>= ' , '1.4.0 ' );
80
80
81
81
$ producerConfig = new Conf ();
82
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
82
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
83
83
$ producer = new Producer ($ producerConfig );
84
84
85
85
$ this ->expectException (Exception::class);
@@ -90,7 +90,7 @@ public function testFromProducerWithMissingConfigShouldFail(): void
90
90
public function testGetBootstraps (): void
91
91
{
92
92
$ clusterConfig = new Conf ();
93
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
93
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
94
94
$ cluster = MockCluster::create (3 , $ clusterConfig );
95
95
96
96
$ bootstrap = $ cluster ->getBootstraps ();
@@ -105,7 +105,7 @@ public function testSetApiVersion(): void
105
105
$ logStack = [];
106
106
107
107
$ clusterConfig = new Conf ();
108
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
108
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
109
109
$ cluster = MockCluster::create (1 , $ clusterConfig );
110
110
111
111
$ cluster ->setApiVersion (ApiKey::Produce, 4 , 6 );
@@ -138,13 +138,13 @@ function (Producer $producer, int $level, string $fac, string $buf) use (&$logSt
138
138
public function testSetPartitionFollowerAndLeader (): void
139
139
{
140
140
$ clusterConfig = new Conf ();
141
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
141
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
142
142
$ cluster = MockCluster::create (2 , $ clusterConfig );
143
143
144
144
$ producerConfig = new Conf ();
145
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
145
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
146
146
$ producerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
147
- $ producerConfig ->set ('topic.metadata.refresh.interval.ms ' , (string ) 50 );
147
+ $ producerConfig ->set ('topic.metadata.refresh.interval.ms ' , (string )50 );
148
148
$ producer = new Producer ($ producerConfig );
149
149
$ producerTopic = $ producer ->newTopic (KAFKA_TEST_TOPIC );
150
150
@@ -168,17 +168,17 @@ public function testSetPartitionFollowerAndLeader(): void
168
168
public function testSetPartitionFollowerWatermarks (): void
169
169
{
170
170
$ clusterConfig = new Conf ();
171
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
171
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
172
172
$ cluster = MockCluster::create (3 , $ clusterConfig );
173
173
174
174
$ cluster ->setPartitionLeader (KAFKA_TEST_TOPIC , 0 , 1 );
175
175
$ cluster ->setPartitionFollower (KAFKA_TEST_TOPIC , 0 , 2 );
176
176
177
177
// produce 10 msgs
178
178
$ producerConfig = new Conf ();
179
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
179
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
180
180
$ producerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
181
- $ producerConfig ->set ('batch.num.messages ' , (string ) 1 );
181
+ $ producerConfig ->set ('batch.num.messages ' , (string )1 );
182
182
$ producer = new Producer ($ producerConfig );
183
183
$ producerTopic = $ producer ->newTopic (KAFKA_TEST_TOPIC );
184
184
for ($ i = 0 ; $ i < 10 ; $ i ++) {
@@ -188,25 +188,23 @@ public function testSetPartitionFollowerWatermarks(): void
188
188
189
189
// prepare consumer
190
190
$ consumerConfig = new Conf ();
191
- $ consumerConfig ->set ('log_level ' , (string ) LOG_EMERG );
191
+ $ consumerConfig ->set ('log_level ' , (string )LOG_EMERG );
192
192
$ consumerConfig ->set ('group.id ' , __METHOD__ );
193
193
$ consumerConfig ->set ('auto.offset.reset ' , 'earliest ' );
194
- $ consumerConfig ->set ('fetch.min.bytes ' , (string ) 100 );
195
- $ consumerConfig ->set ('fetch.message.max.bytes ' , (string ) 1000 );
194
+ $ consumerConfig ->set ('fetch.min.bytes ' , (string )100 );
195
+ $ consumerConfig ->set ('fetch.message.max.bytes ' , (string )1000 );
196
196
$ consumerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
197
197
$ consumer = new KafkaConsumer ($ consumerConfig );
198
198
$ consumer ->assign ([new TopicPartition (KAFKA_TEST_TOPIC , 0 , RD_KAFKA_OFFSET_INVALID )]);
199
199
200
200
// set high watermark to 6
201
201
$ cluster ->setPartitionFollowerWatermarks (KAFKA_TEST_TOPIC , 0 , -1 , 6 );
202
202
203
- sleep (1 );
204
-
205
203
// consume until high watermark
206
204
$ consumedStack = [];
207
205
do {
208
206
$ message = $ consumer ->consume (KAFKA_TEST_SHORT_TIMEOUT_MS );
209
- if ($ message === null ) {
207
+ if ($ message-> err === RD_KAFKA_RESP_ERR__TIMED_OUT && count ( $ consumedStack ) === 0 ) {
210
208
continue ;
211
209
}
212
210
$ consumedStack [] = $ message ->err ?: $ message ->payload ;
@@ -233,13 +231,11 @@ public function testSetPartitionFollowerWatermarks(): void
233
231
// reset high watermark
234
232
$ cluster ->setPartitionFollowerWatermarks (KAFKA_TEST_TOPIC , 0 , -1 , -1 );
235
233
236
- sleep (1 );
237
-
238
234
// consume rest
239
235
$ consumedStack = [];
240
236
do {
241
237
$ message = $ consumer ->consume (KAFKA_TEST_SHORT_TIMEOUT_MS );
242
- if ($ message === null ) {
238
+ if ($ message-> err === RD_KAFKA_RESP_ERR__TIMED_OUT && count ( $ consumedStack ) === 0 ) {
243
239
continue ;
244
240
}
245
241
$ consumedStack [] = $ message ->err ?: $ message ->payload ;
@@ -265,13 +261,13 @@ public function testSetBrokerDownAndUp(): void
265
261
$ this ->requiresLibrdkafkaVersion ('>= ' , '1.4.0 ' );
266
262
267
263
$ clusterConfig = new Conf ();
268
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
264
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
269
265
$ cluster = MockCluster::create (1 , $ clusterConfig );
270
266
271
267
$ producerConfig = new Conf ();
272
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
268
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
273
269
$ producerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
274
- $ producerConfig ->set ('reconnect.backoff.max.ms ' , (string ) 1000 );
270
+ $ producerConfig ->set ('reconnect.backoff.max.ms ' , (string )1000 );
275
271
$ producer = new Producer ($ producerConfig );
276
272
277
273
$ cluster ->setBrokerDown (1 );
@@ -295,12 +291,12 @@ public function testSetBrokerDownAndUp(): void
295
291
public function testPushRequestErrors (): void
296
292
{
297
293
$ clusterConfig = new Conf ();
298
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
294
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
299
295
$ cluster = MockCluster::create (1 , $ clusterConfig );
300
296
301
297
// produce msg
302
298
$ producerConfig = new Conf ();
303
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
299
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
304
300
$ producerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
305
301
$ producer = new Producer ($ producerConfig );
306
302
$ producerTopic = $ producer ->newTopic (KAFKA_TEST_TOPIC );
@@ -317,7 +313,7 @@ public function testPushRequestErrors(): void
317
313
318
314
// try to consume msg
319
315
$ consumerConfig = new Conf ();
320
- $ consumerConfig ->set ('log_level ' , (string ) LOG_EMERG );
316
+ $ consumerConfig ->set ('log_level ' , (string )LOG_EMERG );
321
317
$ consumerConfig ->set ('group.id ' , __METHOD__ );
322
318
// $consumerConfig->set('debug', 'fetch');
323
319
$ consumerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
@@ -337,12 +333,12 @@ public function testPushRequestErrorsArray(): void
337
333
$ this ->requiresLibrdkafkaVersion ('>= ' , '1.7.0 ' );
338
334
339
335
$ clusterConfig = new Conf ();
340
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
336
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
341
337
$ cluster = MockCluster::create (1 , $ clusterConfig );
342
338
343
339
// produce msg
344
340
$ producerConfig = new Conf ();
345
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
341
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
346
342
$ producerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
347
343
$ producer = new Producer ($ producerConfig );
348
344
$ producerTopic = $ producer ->newTopic (KAFKA_TEST_TOPIC );
@@ -361,7 +357,7 @@ public function testPushRequestErrorsArray(): void
361
357
362
358
// try to consume msg
363
359
$ consumerConfig = new Conf ();
364
- $ consumerConfig ->set ('log_level ' , (string ) LOG_EMERG );
360
+ $ consumerConfig ->set ('log_level ' , (string )LOG_EMERG );
365
361
$ consumerConfig ->set ('group.id ' , __METHOD__ );
366
362
// $consumerConfig->set('debug', 'fetch');
367
363
$ consumerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
@@ -381,12 +377,12 @@ public function testCreateTopic(): void
381
377
$ this ->requiresLibrdkafkaVersion ('>= ' , '1.4.0 ' );
382
378
383
379
$ clusterConfig = new Conf ();
384
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
380
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
385
381
$ cluster = MockCluster::create (1 , $ clusterConfig );
386
382
$ cluster ->createTopic (KAFKA_TEST_TOPIC , 12 , 1 );
387
383
388
384
$ producerConfig = new Conf ();
389
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
385
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
390
386
$ producerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
391
387
$ producer = new Producer ($ producerConfig );
392
388
$ producerTopic = $ producer ->newTopic (KAFKA_TEST_TOPIC );
@@ -405,17 +401,17 @@ public function testPushBrokerRequestErrors(): void
405
401
$ this ->requiresLibrdkafkaVersion ('< ' , '1.7.0 ' );
406
402
407
403
$ clusterConfig = new Conf ();
408
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
404
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
409
405
$ cluster = MockCluster::create (1 , $ clusterConfig );
410
406
411
407
// produce msg
412
408
$ producerConfig = new Conf ();
413
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
409
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
414
410
$ producerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
415
411
$ producer = new Producer ($ producerConfig );
416
412
$ producerTopic = $ producer ->newTopic (KAFKA_TEST_TOPIC );
417
413
$ producerTopic ->produce (0 , 0 , __METHOD__ );
418
- $ producer ->flush (KAFKA_TEST_LONG_TIMEOUT_MS );
414
+ $ producer ->flush (KAFKA_TEST_SHORT_TIMEOUT_MS );
419
415
420
416
// first error is retriable, second fatal
421
417
$ cluster ->pushBrokerRequestErrors (
@@ -428,16 +424,16 @@ public function testPushBrokerRequestErrors(): void
428
424
429
425
// try to consume msg
430
426
$ consumerConfig = new Conf ();
431
- $ consumerConfig ->set ('log_level ' , (string ) LOG_EMERG );
427
+ $ consumerConfig ->set ('log_level ' , (string )LOG_EMERG );
432
428
$ consumerConfig ->set ('group.id ' , __METHOD__ );
433
429
// $consumerConfig->set('debug', 'fetch');
434
430
$ consumerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
435
431
$ consumer = new KafkaConsumer ($ consumerConfig );
436
432
$ consumer ->assign ([new TopicPartition (KAFKA_TEST_TOPIC , 0 , rd_kafka_offset_tail (1 ))]);
437
433
438
434
// try to consume msg
439
- $ message1 = $ consumer ->consume (KAFKA_TEST_SHORT_TIMEOUT_MS );
440
- $ message2 = $ consumer ->consume (KAFKA_TEST_SHORT_TIMEOUT_MS );
435
+ $ message1 = $ consumer ->consume (KAFKA_TEST_LONG_TIMEOUT_MS );
436
+ $ message2 = $ consumer ->consume (KAFKA_TEST_LONG_TIMEOUT_MS );
441
437
442
438
$ this ->assertSame (RD_KAFKA_RESP_ERR__AUTHENTICATION , $ message1 ->err );
443
439
$ this ->assertSame (__METHOD__ , $ message2 ->payload );
@@ -448,12 +444,12 @@ public function testPushBrokerRequestErrorRtts(): void
448
444
$ this ->requiresLibrdkafkaVersion ('>= ' , '1.7.0 ' );
449
445
450
446
$ clusterConfig = new Conf ();
451
- $ clusterConfig ->set ('log_level ' , (string ) LOG_EMERG );
447
+ $ clusterConfig ->set ('log_level ' , (string )LOG_EMERG );
452
448
$ cluster = MockCluster::create (1 , $ clusterConfig );
453
449
454
450
// produce msg
455
451
$ producerConfig = new Conf ();
456
- $ producerConfig ->set ('log_level ' , (string ) LOG_EMERG );
452
+ $ producerConfig ->set ('log_level ' , (string )LOG_EMERG );
457
453
$ producerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
458
454
$ producer = new Producer ($ producerConfig );
459
455
$ producerTopic = $ producer ->newTopic (KAFKA_TEST_TOPIC );
@@ -473,7 +469,7 @@ public function testPushBrokerRequestErrorRtts(): void
473
469
474
470
// try to consume msg
475
471
$ consumerConfig = new Conf ();
476
- $ consumerConfig ->set ('log_level ' , (string ) LOG_EMERG );
472
+ $ consumerConfig ->set ('log_level ' , (string )LOG_EMERG );
477
473
$ consumerConfig ->set ('group.id ' , __METHOD__ );
478
474
// $consumerConfig->set('debug', 'fetch');
479
475
$ consumerConfig ->set ('bootstrap.servers ' , $ cluster ->getBootstraps ());
0 commit comments