The distributed consumer in application tests stopped working correctly #386
Description
Ecotone version(s) affected: 1.229.0-latest
Description
We have tests that send a distributed command and check its processing with a delayed retry. After an update, we noticed that tests pass individually, but when running two or more, only one test passes. The remaining tests do not receive the message; it stays in the queue.
How to reproduce
First test for example:
public function testResilientSending(): void
{
$this->logger
->expects(self::exactly(3))
->method('error')
->with(self::isInstanceOf(AggregateNotFoundException::class))
;
$this->distributedBus
->expects(self::once())
->method('convertAndPublishEvent')
->with(
BlockCardHasFailed::ROUTING_KEY,
self::isInstanceOf(BlockCardHasFailed::class),
)
;
$ecotone = $this->bootstrapEcotoneLite();
$ecotone->getDistributedBus()->convertAndSendCommand(
destination: self::SERVICE_NAME,
routingKey: BlockCardCommand::ROUTING_KEY,
command: new BlockCardCommand(
id: new UuidV4(),
)
);
$ecotone->run(
endpointId: self::SERVICE_NAME,
executionPollingMetadata: ExecutionPollingMetadata::createWithFinishWhenNoMessages(false),
);
}
Second test (separate test case):
public function testResilientSending(): void
{
$this->logger
->expects(self::exactly(3))
->method('error')
->with(self::isInstanceOf(AggregateNotFoundException::class))
;
$this->distributedBus
->expects(self::once())
->method('convertAndPublishEvent')
->with(
UnblockCardHasFailed::ROUTING_KEY,
self::isInstanceOf(UnblockCardHasFailed::class),
)
;
$ecotone = $this->bootstrapEcotoneLite();
$ecotone->getDistributedBus()->convertAndSendCommand(
destination: self::SERVICE_NAME,
routingKey: UnblockCardCommand::ROUTING_KEY,
command: new UnblockCardCommand(
id: new UuidV4(),
)
);
$ecotone->run(
endpointId: self::SERVICE_NAME,
executionPollingMetadata: ExecutionPollingMetadata::createWithFinishWhenNoMessages(false),
);
}
UnblockCardTest::testResilientSending
Expectation failed for method name is "error" when invoked 3 time(s).
Method was expected to be called 3 times, actually called 0 times.
EcotoneTestConfiguration:
class EcotoneTestConfiguration
{
#[ServiceContext]
#[Environment(['test'])]
public function errorConfiguration(): ErrorHandlerConfiguration
{
return ErrorHandlerConfiguration::createWithDeadLetterChannel(
errorChannelName: MessagingChannel::Error->value,
delayedRetryTemplate: RetryTemplateBuilder::fixedBackOff(1)
->maxRetryAttempts(2),
deadLetterChannel: MessagingChannel::DeadLetter->value,
);
}
}
private function bootstrapEcotoneLite(): ConfiguredMessagingSystem
{
return EcotoneLite::bootstrap(
classesToResolve: [EcotoneTestConfiguration::class],
containerOrAvailableServices: self::getContainer(),
configuration: ServiceConfiguration::createWithDefaults()
->withEnvironment('test')
->withFailFast(true)
->withDefaultSerializationMediaType(MediaType::APPLICATION_JSON)
->withDefaultErrorChannel(MessagingChannel::Error->value)
->withConsumerMemoryLimit(512)
->withServiceName(self::SERVICE_NAME),
);
}
When each test is run individually, everything works correctly. It also worked fine in all versions up to 1.228.
#[ServiceActivator(inputChannelName: MessagingChannel::Error->value)]
public function handle(
ErrorMessage $message,
#[Header('ecotone_retry_number')]
null|int $retryNumber = null,
): void {
try {
if (!$this->shouldBeSentToDeadLetter($retryNumber)) {
return;
}
$cause = $message->getPayload()->getCause();
if ($cause instanceof Throwable) {
$this->react($cause);
}
} catch (Throwable $e) {
$this->logger->error($e);
}
}
The purpose of the tests is to verify that, after reaching the maximum number of retries with a delay, an error event is sent to other services, while logging each failure at every attempt.