Skip to content

Commit d2f1403

Browse files
authored
improve transaction exceptions (#6)
1 parent 9997ed3 commit d2f1403

5 files changed

+60
-8
lines changed

Diff for: src/Exception/KafkaProducerTransactionAbortException.php

+1-1
Original file line numberDiff line numberDiff line change
@@ -7,5 +7,5 @@
77
class KafkaProducerTransactionAbortException extends \Exception
88
{
99
public const TRANSACTION_REQUIRES_ABORT_EXCEPTION_MESSAGE =
10-
'Produce failed. You need to abort your current transaction and start a new one';
10+
'Produce failed. You need to abort your current transaction and start a new one (%s)';
1111
}

Diff for: src/Exception/KafkaProducerTransactionFatalException.php

+1-1
Original file line numberDiff line numberDiff line change
@@ -7,5 +7,5 @@
77
class KafkaProducerTransactionFatalException extends \Exception
88
{
99
public const FATAL_TRANSACTION_EXCEPTION_MESSAGE =
10-
'Produce failed with a fatal error. This producer instance cannot be used anymore.';
10+
'Produce failed with a fatal error. This producer instance cannot be used anymore (%s)';
1111
}

Diff for: src/Exception/KafkaProducerTransactionRetryException.php

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,5 +6,5 @@
66

77
class KafkaProducerTransactionRetryException extends \Exception
88
{
9-
public const RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE = 'Produce failed but can be retried';
9+
public const RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE = 'Produce failed but can be retried (%s)';
1010
}

Diff for: src/Producer/KafkaProducer.php

+18-3
Original file line numberDiff line numberDiff line change
@@ -255,18 +255,33 @@ private function handleTransactionError(SkcErrorException $e): void
255255
{
256256
if (true === $e->isRetriable()) {
257257
throw new KafkaProducerTransactionRetryException(
258-
KafkaProducerTransactionRetryException::RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE
258+
sprintf(
259+
KafkaProducerTransactionRetryException::RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE,
260+
$e->getMessage()
261+
),
262+
$e->getCode(),
263+
$e
259264
);
260265
} elseif (true === $e->transactionRequiresAbort()) {
261266
throw new KafkaProducerTransactionAbortException(
262-
KafkaProducerTransactionAbortException::TRANSACTION_REQUIRES_ABORT_EXCEPTION_MESSAGE
267+
sprintf(
268+
KafkaProducerTransactionAbortException::TRANSACTION_REQUIRES_ABORT_EXCEPTION_MESSAGE,
269+
$e->getMessage()
270+
),
271+
$e->getCode(),
272+
$e
263273
);
264274
} else {
265275
$this->transactionInitialized = false;
266276
// according to librdkafka documentation, everything that is not retriable, abortable or fatal is fatal
267277
// fatal errors (so stated), need the producer to be destroyed
268278
throw new KafkaProducerTransactionFatalException(
269-
KafkaProducerTransactionFatalException::FATAL_TRANSACTION_EXCEPTION_MESSAGE
279+
sprintf(
280+
KafkaProducerTransactionFatalException::FATAL_TRANSACTION_EXCEPTION_MESSAGE,
281+
$e->getMessage()
282+
),
283+
$e->getCode(),
284+
$e
270285
);
271286
}
272287
}

Diff for: tests/Unit/Producer/KafkaProducerTest.php

+39-2
Original file line numberDiff line numberDiff line change
@@ -368,6 +368,9 @@ public function testBeginTransactionConsecutiveSuccess(): void
368368
public function testBeginTransactionWithRetriableError(): void
369369
{
370370
self::expectException(KafkaProducerTransactionRetryException::class);
371+
self::expectExceptionMessage(
372+
sprintf(KafkaProducerTransactionRetryException::RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE, '')
373+
);
371374

372375
$errorMock = $this->createMock(SkcErrorException::class);
373376
$errorMock->expects(self::once())->method('isRetriable')->willReturn(true);
@@ -389,6 +392,9 @@ public function testBeginTransactionWithRetriableError(): void
389392
public function testBeginTransactionWithAbortError(): void
390393
{
391394
self::expectException(KafkaProducerTransactionAbortException::class);
395+
self::expectExceptionMessage(
396+
sprintf(KafkaProducerTransactionAbortException::TRANSACTION_REQUIRES_ABORT_EXCEPTION_MESSAGE, '')
397+
);
392398

393399
$errorMock = $this->createMock(SkcErrorException::class);
394400
$errorMock->expects(self::once())->method('isRetriable')->willReturn(false);
@@ -411,6 +417,9 @@ public function testBeginTransactionWithAbortError(): void
411417
public function testBeginTransactionWithFatalError(): void
412418
{
413419
self::expectException(KafkaProducerTransactionFatalException::class);
420+
self::expectExceptionMessage(
421+
sprintf(KafkaProducerTransactionFatalException::FATAL_TRANSACTION_EXCEPTION_MESSAGE, '')
422+
);
414423

415424
$errorMock = $this->createMock(SkcErrorException::class);
416425
$errorMock->expects(self::once())->method('isRetriable')->willReturn(false);
@@ -435,6 +444,9 @@ public function testBeginTransactionWithFatalErrorWillTriggerInit(): void
435444
$firstExceptionCaught = false;
436445

437446
self::expectException(KafkaProducerTransactionFatalException::class);
447+
self::expectExceptionMessage(
448+
sprintf(KafkaProducerTransactionFatalException::FATAL_TRANSACTION_EXCEPTION_MESSAGE, '')
449+
);
438450

439451
$errorMock = $this->createMock(SkcErrorException::class);
440452
$errorMock->expects(self::exactly(2))->method('isRetriable')->willReturn(false);
@@ -476,7 +488,9 @@ public function testAbortTransactionSuccess(): void
476488
public function testAbortTransactionFailure(): void
477489
{
478490
self::expectException(KafkaProducerTransactionRetryException::class);
479-
self::expectExceptionMessage(KafkaProducerTransactionRetryException::RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE);
491+
self::expectExceptionMessage(
492+
sprintf(KafkaProducerTransactionRetryException::RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE, 'test')
493+
);
480494

481495
$exception = new SkcErrorException('test', 1, 'some failure', false, true, false);
482496

@@ -507,7 +521,9 @@ public function testCommitTransactionSuccess(): void
507521
public function testCommitTransactionFailure(): void
508522
{
509523
self::expectException(KafkaProducerTransactionRetryException::class);
510-
self::expectExceptionMessage(KafkaProducerTransactionRetryException::RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE);
524+
self::expectExceptionMessage(
525+
sprintf(KafkaProducerTransactionRetryException::RETRIABLE_TRANSACTION_EXCEPTION_MESSAGE, 'test')
526+
);
511527

512528
$exception = new SkcErrorException('test', 1, 'some failure', false, true, false);
513529

@@ -519,4 +535,25 @@ public function testCommitTransactionFailure(): void
519535

520536
$this->kafkaProducer->commitTransaction(10000);
521537
}
538+
539+
/**
540+
* @return void
541+
*/
542+
public function testCommitTransactionFailurePreviousException(): void
543+
{
544+
$exception = new SkcErrorException('test', 1, 'some failure', false, true, false);
545+
546+
$this->rdKafkaProducerMock
547+
->expects(self::once())
548+
->method('commitTransaction')
549+
->with(10000)
550+
->willThrowException($exception);
551+
552+
try {
553+
$this->kafkaProducer->commitTransaction(10000);
554+
} catch (KafkaProducerTransactionRetryException $e) {
555+
self::assertSame($exception, $e->getPrevious());
556+
}
557+
558+
}
522559
}

0 commit comments

Comments
 (0)