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 56785ad

Browse files
committed
add transactional producer
1 parent 3026f0a commit 56785ad

File tree

1 file changed

+149
-0
lines changed

1 file changed

+149
-0
lines changed
Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
<?php
2+
3+
// DISCLAIMER: this feature is not released yet and is subject to change
4+
5+
use Kafka\Configuration;
6+
use Kafka\Message;
7+
use Kafka\Producer;
8+
use Kafka\KafkaErrorException;
9+
10+
error_reporting(E_ALL);
11+
12+
// -------- Intro --------
13+
// The transactional producer operates on top of the idempotent producer,
14+
// and provides full exactly-once semantics (EOS) for Apache Kafka when used
15+
// with the transaction aware consumer (isolation.level=read_committed, which is the default).
16+
17+
// function to handle output of transaction error
18+
function echoTransactionError(KafkaErrorException $e) {
19+
$retryString = 'is not retriable';
20+
$fatalString = 'is not fatal';
21+
$abortString = 'doesn\'t need transaction abort';
22+
23+
if ($e->isFatal()) {
24+
$fatalString = 'is fatal';
25+
}
26+
27+
if ($e->isRetriable()) {
28+
$retryString = 'is retriable';
29+
}
30+
31+
if ($e->transactionRequiresAbort()) {
32+
$abortString = 'needs transaction abort';
33+
}
34+
35+
echo 'Was unable to initialize the transactional producer' . PHP_EOL;
36+
37+
echo sprintf('The reason was: %s, this error %d, %s, %s, %s', $e->getMessage(), $e->getCode(), $fatalString, $retryString, $abortString) . PHP_EOL;
38+
echo sprintf('In detail this means %s', $e->getErrorString()) . PHP_EOL;
39+
echo sprintf('Trace is %s', $e->getTraceAsString()) . PHP_EOL;
40+
}
41+
42+
$conf = new Configuration();
43+
// will be visible in broker logs
44+
$conf->set('client.id', 'pure-php-producer');
45+
// set broker
46+
$conf->set('metadata.broker.list', 'kafka:9096');
47+
// set compression (supported are: none,gzip,lz4,snappy,zstd)
48+
$conf->set('compression.codec', 'snappy');
49+
// set timeout, producer will retry for 5s
50+
$conf->set('message.timeout.ms', '5000');
51+
52+
// For the transactional producer you need a unique id to identify it
53+
$conf->set('transactional.id', 'some-unique-id-of-your-producer-to-recognize-it');
54+
55+
// This callback processes the delivery reports from the broker
56+
// you can see if your message was truly sent
57+
$conf->setDrMsgCb(function (Producer $kafka, Message $message) {
58+
if (RD_KAFKA_RESP_ERR_NO_ERROR !== $message->err) {
59+
$errorStr = rd_kafka_err2str($message->err);
60+
61+
echo sprintf('Message FAILED (%s, %s) to send with payload => %s', $message->err, $errorStr, $message->payload) . PHP_EOL;
62+
} else {
63+
// message successfully delivered
64+
echo sprintf('Message sent SUCCESSFULLY with payload => %s', $message->payload) . PHP_EOL;
65+
}
66+
});
67+
68+
// SASL Authentication
69+
// can be SASL_PLAINTEXT, SASL_SSL
70+
//$conf->set('security.protocol', '');
71+
// can be GSSAPI, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER
72+
// $conf->set('sasl.mechanisms', '');
73+
// $conf->set('sasl.username', '');
74+
// $conf->set('sasl.password', '');
75+
// default is none
76+
// $conf->set('ssl.endpoint.identification.algorithm', 'https');
77+
78+
79+
// SSL Authentication
80+
//$conf->set('security.protocol', 'ssl');
81+
//$conf->set('ssl.ca.location', __DIR__.'/../keys/ca.pem');
82+
//$conf->set('ssl.certificate.location', __DIR__.'/../keys/kafka.cert');
83+
//$conf->set('ssl.key.location', __DIR__.'/../keys/kafka.key');
84+
85+
// Add additional output if you need to debug a problem
86+
// $conf->set('log_level', (string) LOG_DEBUG);
87+
// $conf->set('debug', 'all');
88+
89+
$producer = new Producer($conf);
90+
// initialize producer topic
91+
$topic = $producer->getTopicHandle('pure-php-transactional-test-topic');
92+
// Produce 10 test messages
93+
$amountTestMessages = 10;
94+
95+
// Initialize transactions
96+
try {
97+
$producer->initTransactions(10000);
98+
} catch (KafkaErrorException $e) {
99+
echoTransactionError($e);
100+
die;
101+
}
102+
103+
// Begin transaction for our 10 messages
104+
try {
105+
$producer->beginTransaction();
106+
} catch (KafkaErrorException $e) {
107+
echoTransactionError($e);
108+
die;
109+
}
110+
111+
// Loop to produce some test messages
112+
for ($i = 0; $i < $amountTestMessages; ++$i) {
113+
// Let the partitioner decide the target partition, default partitioner is: RD_KAFKA_MSG_PARTITIONER_CONSISTENT_RANDOM
114+
// You can use a predefined partitioner or write own logic to decide the target partition
115+
$partition = RD_KAFKA_PARTITION_UA;
116+
117+
//produce message with payload, key and headers
118+
$topic->producev(
119+
$partition,
120+
RD_KAFKA_MSG_F_BLOCK, // will block produce if queue is full
121+
sprintf('test message-%d',$i),
122+
sprintf('test-key-%d', $i),
123+
[
124+
'some' => sprintf('header value %d', $i)
125+
]
126+
);
127+
echo sprintf('Queued message number: %d', $i) . PHP_EOL;
128+
129+
// Poll for events e.g. producer callbacks, to handle errors, etc.
130+
// 0 = non-blocking
131+
// -1 = blocking
132+
// any other int value = timeout in ms
133+
$producer->poll(0);
134+
}
135+
136+
// Commit transaction for our 10 messages
137+
try {
138+
$producer->commitTransaction(10000);
139+
} catch (KafkaErrorException $e) {
140+
echoTransactionError($e);
141+
die;
142+
}
143+
144+
// Shutdown producer, flush messages that are in queue. Give up after 20s
145+
$result = $producer->flush(20000);
146+
147+
if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
148+
echo 'Was not able to shutdown within 20s. Messages might be lost!' . PHP_EOL;
149+
}

0 commit comments

Comments
(0)

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