Skip to content

Commit 5addccf

Browse files
nick-zhnick-zh
authored
Support producer transactions (#18)
* save work on producer transactions * update code * save work * up coverage * update readme * update readme * update readme * udpate readme * update readme * update readme
1 parent 3fb81db commit 5addccf

File tree

7 files changed

+410
-0
lines changed

7 files changed

+410
-0
lines changed

‎README.md‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,10 @@ can help out to understand the internals of this library.
1919
- ext-rdkafka: >=4.0.0
2020
- librdkafka: >=0.11.6 (if you use `<librdkafka:1.x` please define your own error callback)
2121

22+
:warning: To use the transactional producer you'll need:
23+
- ext-rdkafka: >=4.1.0
24+
- librdkafka: >=1.4
25+
2226
## Installation
2327
```
2428
composer require jobcloud/php-kafka-lib "~1.0"
@@ -57,6 +61,41 @@ $producer->produce($message);
5761
// Shutdown producer, flush messages that are in queue. Give up after 20s
5862
$result = $producer->flush(20000);
5963
```
64+
65+
##### Transactional producer (needs >=php-rdkafka:4.1 and >=librdkafka:1.4)
66+
```php
67+
<?php
68+
69+
use Jobcloud\Kafka\Message\KafkaProducerMessage;
70+
use Jobcloud\Kafka\Producer\KafkaProducerBuilder;
71+
use Jobcloud\Kafka\Exception\KafkaProducerTransactionRetryException;
72+
use Jobcloud\Kafka\Exception\KafkaProducerTransactionAbortException;
73+
use Jobcloud\Kafka\Exception\KafkaProducerTransactionFatalException;
74+
75+
$producer = KafkaProducerBuilder::create()
76+
->withAdditionalBroker('localhost:9092')
77+
->build();
78+
79+
$message = KafkaProducerMessage::create('test-topic', 0)
80+
->withKey('asdf-asdf-asfd-asdf')
81+
->withBody('some test message payload')
82+
->withHeaders([ 'key' => 'value' ]);
83+
try {
84+
$producer->beginTransaction(10000);
85+
$producer->produce($message);
86+
$producer->commitTransaction(10000);
87+
} catch (KafkaProducerTransactionRetryException $e) {
88+
// something went wrong but you can retry the failed call (either beginTransaction or commitTransaction)
89+
} catch (KafkaProducerTransactionAbortException $e) {
90+
// you need to call $producer->abortTransaction(10000); and try again
91+
} catch (KafkaProducerTransactionFatalException $e) {
92+
// something went very wrong, re-create your producer, otherwise you could jeopardize the idempotency guarantees
93+
}
94+
95+
// Shutdown producer, flush messages that are in queue. Give up after 20s
96+
$result = $producer->flush(20000);
97+
```
98+
6099
##### Avro Producer
61100
To create an avro prodcuer add the avro encoder.
62101

‎src/Exception/KafkaProducerTransactionAbortException.php‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Jobcloud\Kafka\Exception;
6+
7+
class KafkaProducerTransactionAbortException extends \Exception
8+
{
9+
public const TRANSACTION_REQUIRES_ABORT_EXCEPTION_MESSAGE =
10+
'Produce failed. You need to abort your current transaction and start a new one';
11+
}

‎src/Exception/KafkaProducerTransactionFatalException.php‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Jobcloud\Kafka\Exception;
6+
7+
class KafkaProducerTransactionFatalException extends \Exception
8+
{
9+
public const FATAL_TRANSACTION_EXCEPTION_MESSAGE =
10+
'Produce failed with a fatal error. This producer instance cannot be used anymore.';
11+
}

‎src/Exception/KafkaProducerTransactionRetryException.php‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Jobcloud\Kafka\Exception;
6+
7+
class KafkaProducerTransactionRetryException extends \Exception
8+
{
9+
public const RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE = 'Produce failed but can be retried';
10+
}

‎src/Producer/KafkaProducer.php‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,13 +4,17 @@
44

55
namespace Jobcloud\Kafka\Producer;
66

7+
use Jobcloud\Kafka\Exception\KafkaProducerTransactionAbortException;
8+
use Jobcloud\Kafka\Exception\KafkaProducerTransactionFatalException;
9+
use Jobcloud\Kafka\Exception\KafkaProducerTransactionRetryException;
710
use Jobcloud\Kafka\Message\KafkaProducerMessageInterface;
811
use Jobcloud\Kafka\Message\Encoder\EncoderInterface;
912
use Jobcloud\Kafka\Conf\KafkaConfiguration;
1013
use RdKafka\Producer as RdKafkaProducer;
1114
use RdKafka\ProducerTopic as RdKafkaProducerTopic;
1215
use RdKafka\Metadata\Topic as RdKafkaMetadataTopic;
1316
use RdKafka\Exception as RdKafkaException;
17+
use RdKafka\KafkaErrorException as RdKafkaErrorException;
1418

1519
final class KafkaProducer implements KafkaProducerInterface
1620
{
@@ -35,6 +39,11 @@ final class KafkaProducer implements KafkaProducerInterface
3539
*/
3640
protected $encoder;
3741

42+
/**
43+
* @var bool
44+
*/
45+
private $transactionInitialized = false;
46+
3847
/**
3948
* KafkaProducer constructor.
4049
* @param RdKafkaProducer $producer
@@ -160,6 +169,68 @@ public function getMetadataForTopic(string $topicName, int $timeoutMs = 10000):
160169
->current();
161170
}
162171

172+
/**
173+
* Start a producer transaction
174+
*
175+
* @param int $timeoutMs
176+
* @return void
177+
*
178+
* @throws KafkaProducerTransactionAbortException
179+
* @throws KafkaProducerTransactionFatalException
180+
* @throws KafkaProducerTransactionRetryException
181+
*/
182+
public function beginTransaction(int $timeoutMs): void
183+
{
184+
try {
185+
if (false === $this->transactionInitialized) {
186+
$this->producer->initTransactions($timeoutMs);
187+
$this->transactionInitialized = true;
188+
}
189+
190+
$this->producer->beginTransaction();
191+
} catch (RdKafkaErrorException $e) {
192+
$this->handleTransactionError($e);
193+
}
194+
}
195+
196+
/**
197+
* Commit the current producer transaction
198+
*
199+
* @param int $timeoutMs
200+
* @return void
201+
*
202+
* @throws KafkaProducerTransactionAbortException
203+
* @throws KafkaProducerTransactionFatalException
204+
* @throws KafkaProducerTransactionRetryException
205+
*/
206+
public function commitTransaction(int $timeoutMs): void
207+
{
208+
try {
209+
$this->producer->commitTransaction($timeoutMs);
210+
} catch (RdKafkaErrorException $e) {
211+
$this->handleTransactionError($e);
212+
}
213+
}
214+
215+
/**
216+
* Abort the current producer transaction
217+
*
218+
* @param int $timeoutMs
219+
* @return void
220+
*
221+
* @throws KafkaProducerTransactionAbortException
222+
* @throws KafkaProducerTransactionFatalException
223+
* @throws KafkaProducerTransactionRetryException
224+
*/
225+
public function abortTransaction(int $timeoutMs): void
226+
{
227+
try {
228+
$this->producer->abortTransaction($timeoutMs);
229+
} catch (RdKafkaErrorException $e) {
230+
$this->handleTransactionError($e);
231+
}
232+
}
233+
163234
/**
164235
* @param string $topic
165236
* @return RdKafkaProducerTopic
@@ -172,4 +243,31 @@ private function getProducerTopicForTopic(string $topic): RdKafkaProducerTopic
172243

173244
return $this->producerTopics[$topic];
174245
}
246+
247+
/**
248+
* @param RdKafkaErrorException $e
249+
*
250+
* @throws KafkaProducerTransactionAbortException
251+
* @throws KafkaProducerTransactionFatalException
252+
* @throws KafkaProducerTransactionRetryException
253+
*/
254+
private function handleTransactionError(RdKafkaErrorException $e): void
255+
{
256+
if (true === $e->isRetriable()) {
257+
throw new KafkaProducerTransactionRetryException(
258+
KafkaProducerTransactionRetryException::RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE
259+
);
260+
} elseif (true === $e->transactionRequiresAbort()) {
261+
throw new KafkaProducerTransactionAbortException(
262+
KafkaProducerTransactionAbortException::TRANSACTION_REQUIRES_ABORT_EXCEPTION_MESSAGE
263+
);
264+
} else {
265+
$this->transactionInitialized = false;
266+
// according to librdkafka documentation, everything that is not retriable, abortable or fatal is fatal
267+
// fatal errors (so stated), need the producer to be destroyed
268+
throw new KafkaProducerTransactionFatalException(
269+
KafkaProducerTransactionFatalException::FATAL_TRANSACTION_EXCEPTION_MESSAGE
270+
);
271+
}
272+
}
175273
}

‎src/Producer/KafkaProducerInterface.php‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,9 @@
22

33
namespace Jobcloud\Kafka\Producer;
44

5+
use Jobcloud\Kafka\Exception\KafkaProducerTransactionAbortException;
6+
use Jobcloud\Kafka\Exception\KafkaProducerTransactionFatalException;
7+
use Jobcloud\Kafka\Exception\KafkaProducerTransactionRetryException;
58
use Jobcloud\Kafka\Message\KafkaProducerMessageInterface;
69
use RdKafka\Metadata\Topic as RdKafkaMetadataTopic;
710

@@ -71,4 +74,40 @@ public function flush(int $timeoutMs): int;
7174
* @return RdKafkaMetadataTopic
7275
*/
7376
public function getMetadataForTopic(string $topicName, int $timeoutMs = 10000): RdKafkaMetadataTopic;
77+
78+
/**
79+
* Start a producer transaction
80+
*
81+
* @param int $timeoutMs
82+
* @return void
83+
*
84+
* @throws KafkaProducerTransactionAbortException
85+
* @throws KafkaProducerTransactionFatalException
86+
* @throws KafkaProducerTransactionRetryException
87+
*/
88+
public function beginTransaction(int $timeoutMs): void;
89+
90+
/**
91+
* Commit the current producer transaction
92+
*
93+
* @param int $timeoutMs
94+
* @return void
95+
*
96+
* @throws KafkaProducerTransactionAbortException
97+
* @throws KafkaProducerTransactionFatalException
98+
* @throws KafkaProducerTransactionRetryException
99+
*/
100+
public function commitTransaction(int $timeoutMs): void;
101+
102+
/**
103+
* Abort the current producer transaction
104+
*
105+
* @param int $timeoutMs
106+
* @return void
107+
*
108+
* @throws KafkaProducerTransactionAbortException
109+
* @throws KafkaProducerTransactionFatalException
110+
* @throws KafkaProducerTransactionRetryException
111+
*/
112+
public function abortTransaction(int $timeoutMs): void;
74113
}

0 commit comments

Comments
 (0)