diff --git a/packages/Amqp/src/AmqpStreamInboundChannelAdapter.php b/packages/Amqp/src/AmqpStreamInboundChannelAdapter.php index b960c6caf..fd8fbbe44 100644 --- a/packages/Amqp/src/AmqpStreamInboundChannelAdapter.php +++ b/packages/Amqp/src/AmqpStreamInboundChannelAdapter.php @@ -146,7 +146,9 @@ public function receiveWithTimeout(PollingMetadata $pollingMetadata): ?Message $this->startStreamConsuming($context, $pollingMetadata->getEndpointId()); // Wait for messages with the specified timeout - $timeout = $pollingMetadata->getExecutionTimeLimitInMilliseconds() ?: $this->receiveTimeoutInMilliseconds; + $timeout = $pollingMetadata->getExecutionTimeLimitInMilliseconds() > 0 + ? min($pollingMetadata->getExecutionTimeLimitInMilliseconds(), $this->receiveTimeoutInMilliseconds) + : $this->receiveTimeoutInMilliseconds; $timeoutInSeconds = $timeout > 0 ? $timeout / 1000.0 : 10.0; // Keep calling wait() in a loop while the consumer is active diff --git a/packages/Dbal/tests/Integration/Consumer/ExecutionTimeLimitReceiveTimeoutTest.php b/packages/Dbal/tests/Integration/Consumer/ExecutionTimeLimitReceiveTimeoutTest.php new file mode 100644 index 000000000..096ad364b --- /dev/null +++ b/packages/Dbal/tests/Integration/Consumer/ExecutionTimeLimitReceiveTimeoutTest.php @@ -0,0 +1,59 @@ +toRfc4122(); + + $ecotoneLite = EcotoneLite::bootstrapFlowTesting( + containerOrAvailableServices: [ + DbalConnectionFactory::class => $this->getConnectionFactory(), + ], + configuration: ServiceConfiguration::createWithDefaults() + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::DBAL_PACKAGE])) + ->withExtensionObjects([ + DbalBackedMessageChannelBuilder::create($channelName) + ->withReceiveTimeout(200), + ]) + ); + + /** @var PollableChannel $messageChannel */ + $messageChannel = $ecotoneLite->getMessageChannel($channelName); + + $startedAt = microtime(true); + + $receivedMessage = $messageChannel->receiveWithTimeout( + PollingMetadata::create('consumer') + ->setExecutionTimeLimitInMilliseconds(5000) + ); + + $elapsedInMilliseconds = (microtime(true) - $startedAt) * 1000; + + $this->assertNull($receivedMessage); + $this->assertLessThan( + 2000, + $elapsedInMilliseconds, + "Receiving blocked for {$elapsedInMilliseconds}ms - the channel's own receiveTimeout (200ms) was overridden by executionTimeLimit (5000ms), preventing the outer consumer loop from checking termination signals / TimeLimitInterceptor in a timely manner." + ); + } +} diff --git a/packages/Dbal/tests/Integration/Consumer/GracefulShutdownOnSignalTest.php b/packages/Dbal/tests/Integration/Consumer/GracefulShutdownOnSignalTest.php new file mode 100644 index 000000000..e76555167 --- /dev/null +++ b/packages/Dbal/tests/Integration/Consumer/GracefulShutdownOnSignalTest.php @@ -0,0 +1,71 @@ +toRfc4122(); + + // Spawn an independent OS process (not a fork) to deliver the signal - forking + // this PHP process would duplicate the already-open DB connection's socket fd + // (opened in DbalMessagingTestCase::setUp()) into the child and corrupt it. + $parentPid = posix_getpid(); + exec(sprintf('(sleep 0.3 && kill -TERM %d) > /dev/null 2>&1 &', $parentPid)); + + $ecotoneLite = EcotoneLite::bootstrapFlowTesting( + containerOrAvailableServices: [ + DbalConnectionFactory::class => $this->getConnectionFactory(), + ], + configuration: ServiceConfiguration::createWithDefaults() + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::DBAL_PACKAGE])) + ->withExtensionObjects([ + DbalBackedMessageChannelBuilder::create($channelName) + ->withReceiveTimeout(200), + ]) + ); + + $startedAt = microtime(true); + + // Same execution time limit as used in the issue's reproduction steps + $ecotoneLite->run( + $channelName, + ExecutionPollingMetadata::createWithDefaults() + ->withExecutionTimeLimitInMilliseconds(3_600_000) + ); + + $elapsedInMilliseconds = (microtime(true) - $startedAt) * 1000; + + $this->assertLessThan( + 2000, + $elapsedInMilliseconds, + "Consumer kept running for {$elapsedInMilliseconds}ms after receiving SIGTERM instead of shutting down promptly - it was blocked inside a single poll call sized off the 3600000ms executionTimeLimit." + ); + } +} diff --git a/packages/Enqueue/src/EnqueueInboundChannelAdapter.php b/packages/Enqueue/src/EnqueueInboundChannelAdapter.php index c8d0738a1..f66f2de15 100644 --- a/packages/Enqueue/src/EnqueueInboundChannelAdapter.php +++ b/packages/Enqueue/src/EnqueueInboundChannelAdapter.php @@ -57,7 +57,9 @@ public function receiveWithTimeout(PollingMetadata $pollingMetadata): ?Message ); /** @var EnqueueMessage $message */ - $timeoutInMilliseconds = $pollingMetadata->getExecutionTimeLimitInMilliseconds() ?: $this->receiveTimeoutInMilliseconds; + $timeoutInMilliseconds = $pollingMetadata->getExecutionTimeLimitInMilliseconds() > 0 + ? min($pollingMetadata->getExecutionTimeLimitInMilliseconds(), $this->receiveTimeoutInMilliseconds) + : $this->receiveTimeoutInMilliseconds; $message = $consumer->receive($timeoutInMilliseconds); if (! $message) { diff --git a/packages/Kafka/src/Inbound/KafkaInboundChannelAdapter.php b/packages/Kafka/src/Inbound/KafkaInboundChannelAdapter.php index faf12e483..bdfe7f71b 100644 --- a/packages/Kafka/src/Inbound/KafkaInboundChannelAdapter.php +++ b/packages/Kafka/src/Inbound/KafkaInboundChannelAdapter.php @@ -47,7 +47,9 @@ public function receiveWithTimeout(PollingMetadata $pollingMetadata): ?Message ); } - $timeoutInMilliseconds = $pollingMetadata->getExecutionTimeLimitInMilliseconds() ?: $this->receiveTimeoutInMilliseconds; + $timeoutInMilliseconds = $pollingMetadata->getExecutionTimeLimitInMilliseconds() > 0 + ? min($pollingMetadata->getExecutionTimeLimitInMilliseconds(), $this->receiveTimeoutInMilliseconds) + : $this->receiveTimeoutInMilliseconds; if ($timeoutInMilliseconds <= self::MINIMUM_REQUIRED_TIME_FOR_LOAD_BALANCING) { $timeoutInMilliseconds = self::MINIMUM_REQUIRED_TIME_FOR_LOAD_BALANCING; } diff --git a/packages/Kafka/tests/Integration/CommitIntervalTest.php b/packages/Kafka/tests/Integration/CommitIntervalTest.php index 32bfc2698..b9bd93ee9 100644 --- a/packages/Kafka/tests/Integration/CommitIntervalTest.php +++ b/packages/Kafka/tests/Integration/CommitIntervalTest.php @@ -54,7 +54,7 @@ public function test_default_commit_interval_commits_every_message(): void } // Run consumer - should process all 5 - $ecotoneLite->run('kafka_consumer_default', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 5, maxExecutionTimeInMilliseconds: 10000)); + $ecotoneLite->run('kafka_consumer_default', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 5, maxExecutionTimeInMilliseconds: 20000)); $messages = $ecotoneLite->sendQueryWithRouting('consumer.getMessages'); $this->assertCount(5, $messages); @@ -101,12 +101,12 @@ public function test_commit_interval_with_failure_commits_last_successful(): voi } // configuration of commit will be adjusted to single message being consumed - $ecotoneLite->run('kafka_consumer_interval_3_with_failure', ExecutionPollingMetadata::createWithDefaults()->withHandledMessageLimit(15)->withExecutionTimeLimitInMilliseconds(10000)); + $ecotoneLite->run('kafka_consumer_interval_3_with_failure', ExecutionPollingMetadata::createWithDefaults()->withHandledMessageLimit(15)->withExecutionTimeLimitInMilliseconds(20000)); // should only continue and not re-reprocess $ecotoneLite = $this->bootstrapEcotoneLite($topicName, KafkaConsumerWithCommitIntervalAndFailure::class, $consumerInstance); - $ecotoneLite->run('kafka_consumer_interval_3_with_failure', ExecutionPollingMetadata::createWithDefaults()->withHandledMessageLimit(10)->withExecutionTimeLimitInMilliseconds(10000)); + $ecotoneLite->run('kafka_consumer_interval_3_with_failure', ExecutionPollingMetadata::createWithDefaults()->withHandledMessageLimit(10)->withExecutionTimeLimitInMilliseconds(20000)); $this->assertEquals([ 'message_1', 'message_2', 'message_3', 'message_4', 'message_5', diff --git a/packages/Kafka/tests/Integration/FinalFailureStrategyTest.php b/packages/Kafka/tests/Integration/FinalFailureStrategyTest.php index 3b42af789..366a83154 100644 --- a/packages/Kafka/tests/Integration/FinalFailureStrategyTest.php +++ b/packages/Kafka/tests/Integration/FinalFailureStrategyTest.php @@ -100,7 +100,7 @@ public function test_three_messages_second_fails_and_is_released() // Run consumer - should process first message, fail on second, and trigger release $ecotoneTestSupport->run('kafka_channel', ExecutionPollingMetadata::createWithTestingSetup( amountOfMessagesToHandle: 10, - maxExecutionTimeInMilliseconds: 10000, + maxExecutionTimeInMilliseconds: 20000, failAtError: false )); @@ -139,7 +139,7 @@ public function test_three_messages_second_fails_and_is_resend() // Run consumer - should process first message, fail on second, and trigger release $ecotoneTestSupport->run('kafka_channel', ExecutionPollingMetadata::createWithTestingSetup( amountOfMessagesToHandle: 10, - maxExecutionTimeInMilliseconds: 10000, + maxExecutionTimeInMilliseconds: 20000, failAtError: false )); @@ -228,7 +228,7 @@ public function test_when_new_consumer_starts_it_skips_ignored_message() // First run - should process first message (fail and ignore), then process second message (succeed) $ecotoneApp->run('kafka_channel', ExecutionPollingMetadata::createWithTestingSetup( amountOfMessagesToHandle: 2, - maxExecutionTimeInMilliseconds: 10000, + maxExecutionTimeInMilliseconds: 20000, failAtError: false )); @@ -242,7 +242,7 @@ public function test_when_new_consumer_starts_it_skips_ignored_message() $ecotoneApp->run('kafka_channel', ExecutionPollingMetadata::createWithTestingSetup( amountOfMessagesToHandle: 2, - maxExecutionTimeInMilliseconds: 10000, + maxExecutionTimeInMilliseconds: 20000, failAtError: false )); @@ -283,7 +283,7 @@ public function test_different_message_groups_will_handle_ignored_message_twice( // First run - should process first message (fail and ignore), then process second message (succeed) $ecotoneApp->run('kafka_channel', ExecutionPollingMetadata::createWithTestingSetup( amountOfMessagesToHandle: 2, - maxExecutionTimeInMilliseconds: 10000, + maxExecutionTimeInMilliseconds: 20000, failAtError: false )); @@ -315,7 +315,7 @@ public function test_different_message_groups_will_handle_ignored_message_twice( $ecotoneApp->run('kafka_channel', ExecutionPollingMetadata::createWithTestingSetup( amountOfMessagesToHandle: 2, - maxExecutionTimeInMilliseconds: 10000, + maxExecutionTimeInMilliseconds: 20000, failAtError: false )); diff --git a/packages/Kafka/tests/Integration/KafkaMessageChannelTest.php b/packages/Kafka/tests/Integration/KafkaMessageChannelTest.php index b1ae8f9a1..3222880cf 100644 --- a/packages/Kafka/tests/Integration/KafkaMessageChannelTest.php +++ b/packages/Kafka/tests/Integration/KafkaMessageChannelTest.php @@ -650,8 +650,11 @@ public function getConsumed(): array // Both consumers should receive all events independently // Using amountOfMessagesToHandle and maxExecutionTimeInMilliseconds for Kafka consumer group coordination - $consumerService1->run($channelName, ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 10, maxExecutionTimeInMilliseconds: 4000)); - $consumerService2->run($channelName, ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 10, maxExecutionTimeInMilliseconds: 4000)); + // maxExecutionTimeInMilliseconds must comfortably exceed KafkaInboundChannelAdapter::MINIMUM_REQUIRED_TIME_FOR_LOAD_BALANCING (10000ms) + // to leave room for more than one poll cycle - each of these consumers is a brand-new consumer group needing to + // rebalance before it can receive any of the 3 published events. + $consumerService1->run($channelName, ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 10, maxExecutionTimeInMilliseconds: 20000)); + $consumerService2->run($channelName, ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 10, maxExecutionTimeInMilliseconds: 20000)); $this->assertEquals(['event1', 'event2', 'event3'], $consumerService1->sendQueryWithRouting('getConsumed1')); $this->assertEquals(['event1', 'event2', 'event3'], $consumerService2->sendQueryWithRouting('getConsumed2'));