Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Sign up
Appearance settings

Commit 445194e

Browse files
authored
Improve redpanda docker setup (#6)
* add dedicated redpanda example folder * update redpanda examples * update redpanda examples * remove avro deps * update readme * update example readme * update example readme * switch to zstd for now * remove console
1 parent e575eb9 commit 445194e

File tree

13 files changed

+615
-2
lines changed

13 files changed

+615
-2
lines changed

‎README.md

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,9 @@ Checkout these folders to see how to run the examples:
1010
- [php-rdkafka](src/ext-php-rdkafka)
1111
- [php-simple-kafka-client](src/ext-php-simple-kafka-client)
1212

13+
## Examples with other compatible systems
14+
- [redpanda](src/redpanda) Redpanda is a Kafka API compatible streaming platform
15+
1316
## Customize to fit your setup
1417
If you wan't to test / debug something that is closer to your setup,
1518
you can modify the following arguments in `docker-compose.yml`:

‎docker/docker-compose-templates/docker-compose.alpine.redpanda.yml

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,8 @@ services:
77
args:
88
PHP_IMAGE_TAG: 8.0-cli-alpine3.13
99
LIBRDKAFKA_VERSION: v1.6.1
10-
PHP_EXTENSION: arnaud-lb/php-rdkafka
11-
PHP_EXTENSION_VERSION: 5.0.0
10+
PHP_EXTENSION: php-kafka/php-simple-kafka-client
11+
PHP_EXTENSION_VERSION: v0.1.1
1212
tty: true
1313
working_dir: /app
1414
volumes:

‎src/redpanda/README.md

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
# Redpanda playground
2+
[Redpanda](https://vectorized.io/redpanda) support all parts of the Kafka API except for the transactions API.
3+
You can find the issue in their public github [here](https://github.com/vectorizedio/redpanda/issues/445).
4+
Also it seems there is currently no setup to support a schema registry.
5+
6+
## Running examples
7+
1. `docker-compose up -d`
8+
2. `docker-compose exec php composer update`
9+
3. Check the subfolders on how to run the examples

‎src/redpanda/composer.json

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
{
2+
"require": {
3+
"ext-json": "*",
4+
"php-kafka/php-simple-kafka-lib": "dev-main",
5+
"ramsey/uuid": "^4.0"
6+
}
7+
}

‎src/redpanda/docker-compose.yml

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
version: '3.7'
2+
services:
3+
php:
4+
build:
5+
context: ../../docker/php
6+
dockerfile: Dockerfile.alpine
7+
args:
8+
PHP_IMAGE_TAG: 8.0-cli-alpine3.13
9+
LIBRDKAFKA_VERSION: v1.6.1
10+
PHP_EXTENSION: php-kafka/php-simple-kafka-client
11+
PHP_EXTENSION_VERSION: v0.1.1
12+
tty: true
13+
working_dir: /app
14+
volumes:
15+
- ./:/app
16+
redpanda:
17+
entrypoint:
18+
- /usr/bin/rpk
19+
- redpanda
20+
- start
21+
- --smp
22+
- '1'
23+
- --reserve-memory
24+
- 0M
25+
- --overprovisioned
26+
- --node-id
27+
- '0'
28+
- --kafka-addr
29+
- PLAINTEXT://0.0.0.0:29097,OUTSIDE://0.0.0.0:9097
30+
- --advertise-kafka-addr
31+
- PLAINTEXT://redpanda:29097,OUTSIDE://redpanda:9097
32+
- --check=false
33+
image: vectorized/redpanda:v21.4.12
34+
ports:
35+
- 9097:9097
36+
- 29097:29097
Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
# Running consumer / producer
2+
## Prerequisites
3+
Be sure to do this first: [Start containers](./../README.md)
4+
Connect to the php container:
5+
```bash
6+
docker-compose exec php bash
7+
```
8+
9+
## Simple producer
10+
Will per default produce 10 messages:
11+
```bash
12+
php producer.php
13+
```
14+
15+
## Simple consumer
16+
Will consume all messages available:
17+
```bash
18+
php consumer.php
19+
```
Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
<?php
2+
3+
require_once('../vendor/autoload.php');
4+
5+
use PhpKafka\Consumer\KafkaConsumerBuilder;
6+
use PhpKafka\Exception\KafkaConsumerConsumeException;
7+
use PhpKafka\Exception\KafkaConsumerEndOfPartitionException;
8+
use PhpKafka\Exception\KafkaConsumerTimeoutException;
9+
10+
// Get consumer Builder instance
11+
$builder = KafkaConsumerBuilder::create();
12+
13+
// Configure consumer
14+
$consumer = $builder->withAdditionalConfig(
15+
[
16+
// start at the very beginning of the topic when reading for the first time
17+
'auto.offset.reset' => 'earliest',
18+
19+
// will be visible in broker logs
20+
'client.id' => 'php-kafka-lib-high-level-consumer',
21+
22+
// SSL settings
23+
//'security.protocol' => 'ssl',
24+
//'ssl.ca.location' => __DIR__.'/../../../keys/ca.pem',
25+
//'ssl.certificate.location' => __DIR__.'/../../../keys/apl_stage.cert',
26+
//'ssl.key.location' => __DIR__.'/../../../keys/apl_stage.key',
27+
28+
// SASL settings
29+
//'sasl.mechanisms' => '',
30+
//'ssl.endpoint.identification.algorithm' => 'https',
31+
//'sasl.username' => '',
32+
//'sasl.password' => '',
33+
34+
// Add additional output if you need to debug a problem
35+
// 'log_level' => (string) LOG_DEBUG,
36+
// 'debug' => 'all'
37+
]
38+
)
39+
->withAdditionalBroker('redpanda:9097')
40+
->withConsumerGroup('php-kafka-lib-high-level-consumer')
41+
->withSubscription('php-kafka-lib-test-topic')
42+
->build();
43+
44+
$consumer->subscribe();
45+
46+
while (true) {
47+
try {
48+
$message = $consumer->consume(10000);
49+
} catch (KafkaConsumerTimeoutException|KafkaConsumerEndOfPartitionException $e) {
50+
echo 'Didn\'t receive any messages, waiting for more...' . PHP_EOL;
51+
continue;
52+
} catch (KafkaConsumerConsumeException $e) {
53+
echo $e->getMessage() . PHP_EOL;
54+
continue;
55+
}
56+
57+
echo sprintf(
58+
'Read message with key:%s payload:%s topic:%s partition:%d offset:%d headers:%s',
59+
$message->getKey(),
60+
$message->getBody(),
61+
$message->getTopicName(),
62+
$message->getPartition(),
63+
$message->getOffset(),
64+
implode(',', $message->getHeaders())
65+
) . PHP_EOL;
66+
67+
$consumer->commit($message);
68+
}
Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
<?php
2+
3+
require_once('../vendor/autoload.php');
4+
5+
use PhpKafka\Producer\KafkaProducerBuilder;
6+
use PhpKafka\Message\KafkaProducerMessage;
7+
use Ramsey\Uuid\Uuid;
8+
9+
// Get producer Builder instance
10+
$builder = KafkaProducerBuilder::create();
11+
12+
$producer = $builder->withAdditionalConfig(
13+
[
14+
// will be visible in broker logs
15+
'client.id' => 'php-kafka-lib-producer',
16+
// set compression (supported are: none,gzip,lz4,snappy,zstd)
17+
'compression.codec' => 'zstd',
18+
19+
// Add additional output if you need to debug a problem
20+
// 'log_level' => (string) LOG_DEBUG,
21+
// 'debug' => 'all'
22+
]
23+
)
24+
->withAdditionalBroker('redpanda:9097')
25+
->build();
26+
27+
for ($i = 0; $i < 10; ++$i) {
28+
$message = KafkaProducerMessage::create('php-kafka-lib-test-topic', 0)
29+
->withKey(sprintf('test-key-%d', $i))
30+
->withBody(sprintf('test message-%d',$i))
31+
->withHeaders(
32+
[
33+
'some' => 'test header'
34+
]
35+
);
36+
37+
$producer->produce($message);
38+
echo sprintf('Queued message number: %d', $i) . PHP_EOL;
39+
}
40+
41+
// Shutdown producer, flush messages that are in queue. Give up after 20s
42+
$result = $producer->flush(20000);
43+
44+
if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
45+
echo 'Was not able to shutdown within 20s. Messages might be lost!' . PHP_EOL;
46+
}

‎src/redpanda/pure-php/README.md

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
# Running consumer / producer
2+
3+
## Prerequisites
4+
Be sure to do this first: [Start containers](./../README.md)
5+
Connect to the php container:
6+
```bash
7+
docker-compose exec php bash
8+
```
9+
10+
## Simple producer
11+
Will per default produce 10 messages:
12+
```bash
13+
cd pure-php
14+
php producer.php
15+
```
16+
17+
## Transactional producer (currently unsupported by Redpanda)
18+
Will per default produce 10 messages:
19+
```bash
20+
cd pure-php
21+
php producer_transactional.php
22+
```
23+
24+
## Consumer
25+
Will consume all messages available:
26+
```bash
27+
cd pure-php
28+
php consumer.php
29+
```
30+
31+
## Query metadata
32+
Will query metadata for all available topics, etc.:
33+
```bash
34+
cd pure-php
35+
php metadata.php
36+
```

‎src/redpanda/pure-php/consumer.php

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
<?php
2+
3+
use SimpleKafkaClient\Configuration;
4+
use SimpleKafkaClient\Consumer;
5+
6+
$conf = new Configuration();
7+
// will be visible in broker logs
8+
$conf->set('client.id', 'pure-php-high-level-consumer');
9+
// set consumer group, e.g. <my-application-name>-consumer
10+
$conf->set('group.id', 'pure-php-high-level-consumer');
11+
// set broker
12+
$conf->set('metadata.broker.list', 'redpanda:9097');
13+
// don't auto commit, give the application the control to do that (default is: true)
14+
$conf->set('enable.auto.commit', 'false');
15+
// start at the very beginning of the topic when reading for the first time
16+
$conf->set('auto.offset.reset', 'earliest');
17+
// Get eof code instead of null
18+
$conf->set('enable.partition.eof', 'true');
19+
20+
// SASL Authentication
21+
//$conf->set('sasl.mechanisms', '');
22+
//$conf->set('ssl.endpoint.identification.algorithm', 'https');
23+
//$conf->set('sasl.username', '');
24+
//$conf->set('sasl.password', '');
25+
26+
// SSL Authentication
27+
//$conf->set('security.protocol', 'ssl');
28+
//$conf->set('ssl.ca.location', __DIR__.'/../../../keys/ca.pem');
29+
//$conf->set('ssl.certificate.location', __DIR__.'/../../../keys/kafka.cert');
30+
//$conf->set('ssl.key.location', __DIR__.'/../../../keys/kafka.key');
31+
32+
// Add additional output if you need to debug a problem
33+
// $conf->set('log_level', (string) LOG_DEBUG);
34+
// $conf->set('debug', 'all');
35+
36+
// Create high level consumer
37+
$consumer = new Consumer($conf);
38+
39+
// Subscribe to one or multiple topics
40+
$consumer->subscribe(['pure-php-test-topic', 'pure-php-transactional-test-topic']);
41+
42+
while (true) {
43+
// Try to consume messages for the given timout (20s)
44+
$message = $consumer->consume(20000);
45+
46+
if (RD_KAFKA_RESP_ERR__PARTITION_EOF === $message->err) {
47+
echo 'Reached end of partition, waiting for more messages...' . PHP_EOL;
48+
continue;
49+
} else if (RD_KAFKA_RESP_ERR__TIMED_OUT === $message->err) {
50+
echo 'Timed out without receiving a new message, waiting for more messages...' . PHP_EOL;
51+
continue;
52+
} else if (RD_KAFKA_RESP_ERR_NO_ERROR !== $message->err) {
53+
echo kafka_err2str($message->err) . PHP_EOL;
54+
continue;
55+
}
56+
57+
echo sprintf(
58+
'Read message with key:%s payload:%s topic:%s partition:%d offset:%d',
59+
$message->key,
60+
$message->payload,
61+
$message->topic_name,
62+
$message->partition,
63+
$message->offset
64+
) . PHP_EOL;
65+
// Here is where you do your business logic to process your message
66+
// after you have done so, commit the message offset to the broker
67+
68+
// commit the message(s) offset synchronous back to the broker
69+
$consumer->commit($message);
70+
71+
// you can also commit the message(s) offset in an async manner, which is slightly faster
72+
// but poses of course the challenge of handling errors in an async manner as well
73+
//$consumer->commitAsync($message);
74+
}

0 commit comments

Comments
(0)

AltStyle によって変換されたページ (->オリジナル) /