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 8d47340

Browse files
authored
add avro example for redpanda (#11)
1 parent 7c62d87 commit 8d47340

File tree

10 files changed

+292
-2
lines changed

10 files changed

+292
-2
lines changed

‎src/ext-php-simple-kafka-client/php-simple-kafka-lib/avroSchema/nickzh.php.kafka.examples.entity.product-value.avsc

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
{
2626
"name": "price",
2727
"type": "float",
28-
"default": null,
28+
"default": 0.0,
2929
"doc": "Price of our product"
3030
}
3131
]

‎src/redpanda/composer.json

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
{
22
"require": {
33
"ext-json": "*",
4+
"flix-tech/avro-serde-php": "^1.4",
5+
"jobcloud/php-console-kafka-schema-registry": "^1.1",
46
"php-kafka/php-simple-kafka-lib": "dev-main",
57
"ramsey/uuid": "^4.0"
68
}

‎src/redpanda/docker-compose.yml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,4 +33,5 @@ services:
3333
image: vectorized/redpanda:v21.10.2
3434
ports:
3535
- 9097:9097
36+
- 8081:8081
3637
- 29097:29097

‎src/redpanda/php-simple-kafka-lib/README.md

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,10 @@
1+
# Register test schema
2+
Run the following to register the test schema:
3+
```bash
4+
cd php-simple-kafka-lib
5+
./console kafka-schema-registry:register:changed avroSchema
6+
```
7+
18
# Running consumer / producer
29
## Prerequisites
310
Be sure to do this first: [Start containers](./../README.md)
Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
<?php
2+
3+
require_once('../vendor/autoload.php');
4+
5+
use FlixTech\AvroSerializer\Objects\RecordSerializer;
6+
use FlixTech\SchemaRegistryApi\Exception\SchemaNotFoundException;
7+
use FlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
8+
use FlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
9+
use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
10+
use FlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
11+
use GuzzleHttp\Client;
12+
use PhpKafka\Message\Decoder\AvroDecoder;
13+
use PhpKafka\Message\Registry\AvroSchemaRegistry;
14+
use PhpKafka\Message\KafkaAvroSchema;
15+
use PhpKafka\Consumer\KafkaConsumerBuilder;
16+
use PhpKafka\Exception\KafkaConsumerTimeoutException;
17+
use PhpKafka\Exception\KafkaConsumerEndOfPartitionException;
18+
use PhpKafka\Exception\KafkaConsumerConsumeException;
19+
20+
// Instantiate cached schema registry (vendor: flix)
21+
$registry = new CachedRegistry(
22+
new BlockingRegistry(
23+
new PromisingRegistry(
24+
new Client(
25+
[
26+
'base_uri' => 'http://redpanda:8081',
27+
//'auth' => ['user', 'pw']
28+
]
29+
)
30+
)
31+
),
32+
new AvroObjectCacheAdapter()
33+
);
34+
35+
// Instantiate schema registry of lib (Note: In the future we will use our won cached registry)
36+
$schemaRegistry = new AvroSchemaRegistry($registry);
37+
// add schema for topic
38+
$schemaRegistry->addBodySchemaMappingForTopic(
39+
'php-kafka-lib-test-topic-avro',
40+
new KafkaAvroSchema(
41+
'nickzh.php.kafka.examples.entity.product-value'
42+
// optional param - version: if not passed we will take latest
43+
)
44+
);
45+
$schemaRegistry->addKeySchemaMappingForTopic(
46+
'php-kafka-lib-test-topic-avro',
47+
new KafkaAvroSchema(
48+
'nickzh.php.kafka.examples.entity.product-key'
49+
// optional param - version: if not passed we will take latest
50+
)
51+
);
52+
53+
// instantiate avro record serializer (vendor: flix)
54+
$recordSerializer = new RecordSerializer($registry);
55+
56+
// initialize Avro decoder (Note: In the future, we will use our own record serializer)
57+
$decoder = new AvroDecoder($schemaRegistry, $recordSerializer);
58+
59+
// Get consumer Builder instance
60+
$builder = KafkaConsumerBuilder::create();
61+
62+
// Configure consumer
63+
$consumer = $builder->withAdditionalConfig(
64+
[
65+
// start at the very beginning of the topic when reading for the first time
66+
'auto.offset.reset' => 'earliest',
67+
68+
// will be visible in broker logs
69+
'client.id' => 'php-kafka-lib-high-level-consumer-avro',
70+
71+
// SSL settings
72+
//'security.protocol' => 'ssl',
73+
//'ssl.ca.location' => __DIR__.'/../../../keys/ca.pem',
74+
//'ssl.certificate.location' => __DIR__.'/../../../keys/apl_stage.cert',
75+
//'ssl.key.location' => __DIR__.'/../../../keys/apl_stage.key',
76+
77+
// SASL settings
78+
//'sasl.mechanisms' => '',
79+
//'ssl.endpoint.identification.algorithm' => 'https',
80+
//'sasl.username' => '',
81+
//'sasl.password' => '',
82+
83+
// Add additional output if you need to debug a problem
84+
// 'log_level' => (string) LOG_DEBUG,
85+
// 'debug' => 'all'
86+
]
87+
)
88+
->withAdditionalBroker('redpanda:9097')
89+
->withConsumerGroup('php-kafka-lib-high-level-consumer-avro')
90+
->withDecoder($decoder)
91+
->withSubscription('php-kafka-lib-test-topic-avro')
92+
->build();
93+
94+
$consumer->subscribe();
95+
96+
while (true) {
97+
try {
98+
$message = $consumer->consume(10000);
99+
} catch (KafkaConsumerTimeoutException|KafkaConsumerEndOfPartitionException $e) {
100+
continue;
101+
} catch (KafkaConsumerConsumeException $e) {
102+
echo $e->getMessage() . PHP_EOL;
103+
continue;
104+
} catch (SchemaNotFoundException $e) {
105+
echo 'Consumed message with no or unknown schema' . PHP_EOL;
106+
continue;
107+
}
108+
109+
echo sprintf(
110+
'Read message with key:%s payload:%s topic:%s partition:%d offset:%d headers:%s',
111+
$message->getKey(),
112+
implode(',', $message->getBody()),
113+
$message->getTopicName(),
114+
$message->getPartition(),
115+
$message->getOffset(),
116+
implode(',', $message->getHeaders())
117+
) . PHP_EOL;
118+
119+
$consumer->commit($message);
120+
}
Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
<?php
2+
3+
require_once('../vendor/autoload.php');
4+
5+
use FlixTech\AvroSerializer\Objects\RecordSerializer;
6+
use FlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
7+
use FlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
8+
use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
9+
use FlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
10+
use GuzzleHttp\Client;
11+
use PhpKafka\Message\Encoder\AvroEncoder;
12+
use PhpKafka\Message\KafkaAvroSchema;
13+
use PhpKafka\Message\Registry\AvroSchemaRegistry;
14+
use PhpKafka\Producer\KafkaProducerBuilder;
15+
use PhpKafka\Message\KafkaProducerMessage;
16+
use Ramsey\Uuid\Uuid;
17+
18+
// Instantiate cached schema registry (vendor: flix)
19+
$registry = new CachedRegistry(
20+
new BlockingRegistry(
21+
new PromisingRegistry(
22+
new Client(
23+
[
24+
'base_uri' => 'http://redpanda:8081',
25+
//'auth' => ['user', 'pw']
26+
]
27+
)
28+
)
29+
),
30+
new AvroObjectCacheAdapter()
31+
);
32+
33+
// Instantiate schema registry of lib (Note: In the future we will use our won cached registry)
34+
$schemaRegistry = new AvroSchemaRegistry($registry);
35+
// add schema for topic
36+
$schemaRegistry->addBodySchemaMappingForTopic(
37+
'php-kafka-lib-test-topic-avro',
38+
new KafkaAvroSchema(
39+
'nickzh.php.kafka.examples.entity.product-value'
40+
// optional param - version: if not passed we will take latest
41+
)
42+
);
43+
$schemaRegistry->addKeySchemaMappingForTopic(
44+
'php-kafka-lib-test-topic-avro',
45+
new KafkaAvroSchema(
46+
'nickzh.php.kafka.examples.entity.product-key'
47+
// optional param - version: if not passed we will take latest
48+
)
49+
);
50+
51+
// instantiate avro record serializer (vendor: flix)
52+
$recordSerializer = new RecordSerializer($registry);
53+
54+
// initialize Avro encode (Note: In the future, we will use our own record serializer)
55+
$encoder = new AvroEncoder($schemaRegistry, $recordSerializer);
56+
57+
// Get producer Builder instance
58+
$builder = KafkaProducerBuilder::create();
59+
60+
$producer = $builder->withAdditionalConfig(
61+
[
62+
// will be visible in broker logs
63+
'client.id' => 'php-kafka-lib-producer-avro',
64+
// set compression (supported are: none,gzip,lz4,snappy,zstd)
65+
'compression.codec' => 'snappy',
66+
67+
// Add additional output if you need to debug a problem
68+
// 'log_level' => (string) LOG_DEBUG,
69+
// 'debug' => 'all'
70+
]
71+
)
72+
->withAdditionalBroker('redpanda:9097')
73+
->withEncoder($encoder)
74+
->build();
75+
76+
for ($i = 0; $i < 10; ++$i) {
77+
$message = KafkaProducerMessage::create('php-kafka-lib-test-topic-avro', 0)
78+
->withKey(sprintf('test-key-%d', $i))
79+
->withBody(
80+
[
81+
'id' => Uuid::uuid6()->toString(),
82+
'name' => sprintf('Product %d', $i),
83+
'description' => 'A random test product',
84+
'price' => 21.25
85+
]
86+
)
87+
->withHeaders(
88+
[
89+
'some' => 'test header'
90+
]
91+
);
92+
93+
$producer->produce($message);
94+
echo sprintf('Queued message number: %d', $i) . PHP_EOL;
95+
}
96+
97+
// Shutdown producer, flush messages that are in queue. Give up after 20s
98+
$result = $producer->flush(20000);
99+
100+
if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
101+
echo 'Was not able to shutdown within 20s. Messages might be lost!' . PHP_EOL;
102+
}
Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
{
2+
"type": "string"
3+
}
Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
{
2+
"type": "record",
3+
"name": "product",
4+
"namespace": "nickzh.php.kafka.examples.entity",
5+
"fields": [
6+
{
7+
"name": "id",
8+
"type": "string",
9+
"doc": "UUID of our product"
10+
},
11+
{
12+
"name": "name",
13+
"type": "string",
14+
"doc": "Name of our product"
15+
},
16+
{
17+
"name": "description",
18+
"type": [
19+
"null",
20+
"string"
21+
],
22+
"default": null,
23+
"doc": "Description of our product"
24+
},
25+
{
26+
"name": "price",
27+
"type": "float",
28+
"default": 0.0,
29+
"doc": "Price of our product"
30+
}
31+
]
32+
}
Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
#!/usr/bin/env php
2+
<?php
3+
declare(strict_types=1);
4+
5+
namespace Jobcloud\Message\Schema;
6+
7+
include_once __DIR__ . '/../vendor/autoload.php';
8+
9+
use Jobcloud\Kafka\SchemaRegistryClient\ServiceProvider\KafkaSchemaRegistryApiClientProvider;
10+
use Symfony\Component\Console\Application;
11+
use Jobcloud\SchemaConsole\ServiceProvider\CommandServiceProvider;
12+
use Pimple\Container;
13+
14+
$container = new Container();
15+
16+
$settings[KafkaSchemaRegistryApiClientProvider::SETTING_KEY_BASE_URL] = 'http://redpanda:8081';
17+
$container->offsetSet(KafkaSchemaRegistryApiClientProvider::CONTAINER_KEY, $settings);
18+
19+
$container->register(new CommandServiceProvider());
20+
21+
$application = new Application();
22+
$application->addCommands($container[CommandServiceProvider::COMMANDS]);
23+
$application->run();

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
use SimpleKafkaClient\Producer;
66
use SimpleKafkaClient\KafkaErrorException;
77

8-
echo 'Transaction support not yet working, see: https://github.com/vectorizedio/redpanda/issues/445' . PHP_EOL;die;
8+
echo 'Transaction example is currently work in progress, see https://github.com/php-kafka/php-kafka-examples/issues/10' . PHP_EOL;die;
99

1010
error_reporting(E_ALL);
1111

0 commit comments

Comments
(0)

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