|
16 | 16 | $conf->set('compression.codec', 'snappy'); |
17 | 17 | // set timeout, producer will retry for 5s |
18 | 18 | $conf->set('message.timeout.ms', '5000'); |
| 19 | +$conf->set('linger.ms', '1000'); |
19 | 20 | //If you need to produce exactly once and want to keep the original produce order, uncomment the line below |
20 | 21 | //$conf->set('enable.idempotence', 'true'); |
21 | 22 |
|
|
28 | 29 | echo sprintf('Message FAILED (%s, %s) to send with payload => %s', $message->err, $errorStr, $message->payload) . PHP_EOL; |
29 | 30 | } else { |
30 | 31 | // message successfully delivered |
31 | | - echo sprintf('Message sent SUCCESSFULLY with payload => %s', $message->payload) . PHP_EOL; |
| 32 | + //echo sprintf('Message sent SUCCESSFULLY with payload => %s', 'x') . PHP_EOL; |
32 | 33 | } |
33 | 34 | }); |
34 | 35 |
|
|
57 | 58 | // initialize producer topic |
58 | 59 | $topic = $producer->newTopic('pure-php-test-topic'); |
59 | 60 | // Produce 10 test messages |
60 | | -$amountTestMessages = 10; |
| 61 | +$amountTestMessages = 1000000; |
61 | 62 |
|
62 | 63 | // Loop to produce some test messages |
63 | 64 | for ($i = 0; $i < $amountTestMessages; ++$i) { |
|
69 | 70 | $topic->producev( |
70 | 71 | $partition, |
71 | 72 | RD_KAFKA_MSG_F_BLOCK, // will block produce if queue is full |
72 | | - sprintf('test message-%d',$i), |
| 73 | + //sprintf('test message-%d',$i), |
| 74 | + str_repeat('12312313123213123',10000), |
73 | 75 | sprintf('test-key-%d', $i), |
74 | 76 | [ |
75 | 77 | 'some' => sprintf('header value %d', $i) |
76 | 78 | ] |
77 | 79 | ); |
78 | | - echo sprintf('Queued message number: %d', $i) . PHP_EOL; |
| 80 | + if ($i%1000 === 0) { |
| 81 | + echo sprintf('Queued message number: %d', $i) . PHP_EOL; |
| 82 | + } |
| 83 | + if ($i%5000 === 0) { |
| 84 | + $producer->flush(10000); |
| 85 | + } |
79 | 86 |
|
80 | 87 | // Poll for events e.g. producer callbacks, to handle errors, etc. |
81 | 88 | // 0 = non-blocking |
|
0 commit comments