ecotoneframework · dgafka · Jul 2, 2026 · Jul 2, 2026
diff --git 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
@@ -0,0 +1,59 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Integration\Consumer;
+
+use Ecotone\Dbal\DbalBackedMessageChannelBuilder;
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Messaging\Config\ModulePackageList;
+use Ecotone\Messaging\Config\ServiceConfiguration;
+use Ecotone\Messaging\Endpoint\PollingMetadata;
+use Ecotone\Messaging\PollableChannel;
+use Enqueue\Dbal\DbalConnectionFactory;
+use Symfony\Component\Uid\Uuid;
+use Test\Ecotone\Dbal\DbalMessagingTestCase;
+
+/**
+ * Reproduces https://github.com/ecotoneframework/ecotone-dev/issues/440
+ *
+ * @internal
+ */
+final class ExecutionTimeLimitReceiveTimeoutTest extends DbalMessagingTestCase
+{
+    public function test_execution_time_limit_should_not_override_the_configured_receive_timeout_on_empty_queue(): void
+    {
+        $channelName = Uuid::v7()->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
@@ -0,0 +1,71 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Integration\Consumer;
+
+use Ecotone\Dbal\DbalBackedMessageChannelBuilder;
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Messaging\Config\ModulePackageList;
+use Ecotone\Messaging\Config\ServiceConfiguration;
+use Ecotone\Messaging\Endpoint\ExecutionPollingMetadata;
+use Enqueue\Dbal\DbalConnectionFactory;
+use Symfony\Component\Uid\Uuid;
+use Test\Ecotone\Dbal\DbalMessagingTestCase;
+
+/**
+ * Reproduces https://github.com/ecotoneframework/ecotone-dev/issues/440
+ *
+ * Mirrors the exact repro steps from the issue: start a consumer with a large
+ * executionTimeLimit (e.g. `--executionTimeLimit 3600000`), send it SIGTERM while
+ * it's idle-polling an empty queue, and expect it to shut down promptly instead of
+ * ignoring the signal until the execution time limit elapses.
+ *
+ * @internal
+ */
+final class GracefulShutdownOnSignalTest extends DbalMessagingTestCase
+{
+    public function test_consumer_stops_promptly_on_sigterm_despite_large_execution_time_limit(): void
+    {
+        if (! extension_loaded('pcntl') || ! extension_loaded('posix')) {
+            self::markTestSkipped('pcntl and posix extensions are required to send a real termination signal.');
+        }
+
+        $channelName = Uuid::v7()->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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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'));