diff --git a/Monorepo/Benchmark/AsyncPublishingBenchmark.php b/Monorepo/Benchmark/AsyncPublishingBenchmark.php
index 63a00ca13..4814fc83b 100644
--- a/Monorepo/Benchmark/AsyncPublishingBenchmark.php
+++ b/Monorepo/Benchmark/AsyncPublishingBenchmark.php
@@ -72,7 +72,7 @@ public function setUpAmqpBatchChannel(): void
     {
         $this->batchChannel = $this->bootstrapBatchChannel(
             ModulePackageList::AMQP_PACKAGE,
-            AmqpBackedMessageChannelBuilder::create(uniqid('benchmark_orders_'))->withAsyncPublishing(),
+            AmqpBackedMessageChannelBuilder::create(uniqid('benchmark_orders_'))->withHighThroughputPublishing(),
             [AmqpConnectionFactory::class => new AmqpConnectionFactory(['dsn' => getenv('RABBIT_HOST') ?: 'amqp://guest:guest@localhost:5672/%2f'])],
         );
         $this->warmUpBatchChannel();
@@ -95,7 +95,7 @@ public function setUpKafkaBatchChannel(): void
         $uniqueId = uniqid('benchmark_orders_');
         $this->batchChannel = $this->bootstrapBatchChannel(
             ModulePackageList::KAFKA_PACKAGE,
-            KafkaMessageChannelBuilder::create($uniqueId, topicName: $uniqueId, messageGroupId: $uniqueId)->withAsyncPublishing(),
+            KafkaMessageChannelBuilder::create($uniqueId, topicName: $uniqueId, messageGroupId: $uniqueId)->withHighThroughputPublishing(),
             [KafkaBrokerConfiguration::class => KafkaBrokerConfiguration::createWithDefaults([getenv('KAFKA_DSN') ?: 'localhost:9094'])],
         );
         $this->warmUpBatchChannel();
@@ -111,7 +111,7 @@ public function setUpDbalBatchChannel(): void
     {
         $this->batchChannel = $this->bootstrapBatchChannel(
             ModulePackageList::DBAL_PACKAGE,
-            DbalBackedMessageChannelBuilder::create(uniqid('benchmark_orders_'))->withAsyncPublishing(),
+            DbalBackedMessageChannelBuilder::create(uniqid('benchmark_orders_'))->withHighThroughputPublishing(),
             [DbalConnectionFactory::class => new DbalConnectionFactory(getenv('DATABASE_DSN') ?: 'pgsql://ecotone:secret@localhost:5432/ecotone')],
         );
         $this->warmUpBatchChannel();
@@ -127,7 +127,7 @@ public function setUpRedisBatchChannel(): void
     {
         $this->batchChannel = $this->bootstrapBatchChannel(
             ModulePackageList::REDIS_PACKAGE,
-            RedisBackedMessageChannelBuilder::create(uniqid('benchmark_orders_'))->withAsyncPublishing(),
+            RedisBackedMessageChannelBuilder::create(uniqid('benchmark_orders_'))->withHighThroughputPublishing(),
             [RedisConnectionFactory::class => new RedisConnectionFactory(getenv('REDIS_DSN') ?: 'redis://localhost:6379')],
         );
         $this->warmUpBatchChannel();
@@ -149,7 +149,7 @@ public function setUpSqsBatchChannel(): void
     {
         $this->batchChannel = $this->bootstrapBatchChannel(
             ModulePackageList::SQS_PACKAGE,
-            SqsBackedMessageChannelBuilder::create(uniqid('benchmark_orders_'))->withAsyncPublishing(),
+            SqsBackedMessageChannelBuilder::create(uniqid('benchmark_orders_'))->withHighThroughputPublishing(),
             [SqsConnectionFactory::class => new SqsConnectionFactory(getenv('SQS_DSN') ?: 'sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localhost:4566&version=latest')],
         );
         $this->warmUpBatchChannel();
diff --git a/Monorepo/Benchmark/OutboxRelayBenchmark.php b/Monorepo/Benchmark/OutboxRelayBenchmark.php
new file mode 100644
index 000000000..61b314b89
--- /dev/null
+++ b/Monorepo/Benchmark/OutboxRelayBenchmark.php
@@ -0,0 +1,248 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Monorepo\Benchmark;
+
+use Ecotone\Amqp\AmqpBackedMessageChannelBuilder;
+use Ecotone\Dbal\DbalBackedMessageChannelBuilder;
+use Ecotone\Dbal\OutboxForwardingMessageChannel;
+use Ecotone\Kafka\Channel\KafkaMessageChannelBuilder;
+use Ecotone\Kafka\Configuration\KafkaBrokerConfiguration;
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Lite\Test\FlowTestSupport;
+use Ecotone\Messaging\Attribute\Asynchronous;
+use Ecotone\Messaging\Channel\CombinedMessageChannel;
+use Ecotone\Messaging\Channel\SimpleMessageChannelBuilder;
+use Ecotone\Messaging\Config\ModulePackageList;
+use Ecotone\Messaging\Config\ServiceConfiguration;
+use Ecotone\Messaging\Endpoint\ExecutionPollingMetadata;
+use Ecotone\Modelling\Attribute\CommandHandler;
+use Ecotone\Redis\RedisBackedMessageChannelBuilder;
+use Ecotone\Sqs\SqsBackedMessageChannelBuilder;
+use Ecotone\Test\LicenceTesting;
+use Enqueue\AmqpExt\AmqpConnectionFactory;
+use Enqueue\Dbal\DbalConnectionFactory;
+use Enqueue\Redis\RedisConnectionFactory;
+use Enqueue\Sqs\SqsConnectionFactory;
+use PhpBench\Attributes\BeforeMethods;
+use PhpBench\Attributes\Iterations;
+use PhpBench\Attributes\Revs;
+use PhpBench\Attributes\Warmup;
+
+/**
+ * Measures how fast the whole DBAL outbox is drained and handed over to the next channel of a combined channel:
+ * message-by-message forwarding (no enterprise licence) against batched SQL drain-and-forward (enterprise).
+ * The consumer is warmed up before messages are published, so only steady-state relay work is measured.
+ * The in-memory target subjects isolate the producing side of the relay; the provider subjects show the full
+ * path into real brokers receiving whole batches at once.
+ */
+#[Warmup(0), Revs(1), Iterations(5)]
+class OutboxRelayBenchmark
+{
+    private const AMOUNT_OF_RELAYED_MESSAGES = 10_000;
+
+    private const MESSAGE_PAYLOAD = 'benchmark order payload for outbox relay comparison';
+
+    private FlowTestSupport $messaging;
+
+    public function setUpRelayMessageByMessage(): void
+    {
+        $this->messaging = $this->bootstrapOutbox(licenceKey: null);
+        $this->warmUpConsumer();
+        $this->fillOutbox();
+    }
+
+    public function setUpRelayBatched(): void
+    {
+        $this->messaging = $this->bootstrapOutbox(licenceKey: LicenceTesting::VALID_LICENCE);
+        $this->warmUpConsumer();
+        $this->fillOutbox();
+    }
+
+    public function setUpRelaySingleBatch(): void
+    {
+        $this->messaging = $this->bootstrapOutbox(licenceKey: LicenceTesting::VALID_LICENCE, maxForwardingBatchSize: self::AMOUNT_OF_RELAYED_MESSAGES);
+        $this->warmUpConsumer();
+        $this->fillOutbox();
+    }
+
+    public function setUpRelayBatchedIntoDbalTarget(): void
+    {
+        $this->messaging = $this->bootstrapOutbox(licenceKey: LicenceTesting::VALID_LICENCE, targetProvider: 'dbal');
+        $this->warmUpConsumer();
+        $this->fillOutbox();
+    }
+
+    public function setUpRelayBatchedIntoAmqpTarget(): void
+    {
+        $this->messaging = $this->bootstrapOutbox(licenceKey: LicenceTesting::VALID_LICENCE, targetProvider: 'amqp');
+        $this->warmUpConsumer();
+        $this->fillOutbox();
+    }
+
+    public function setUpRelayBatchedIntoKafkaTarget(): void
+    {
+        $this->messaging = $this->bootstrapOutbox(licenceKey: LicenceTesting::VALID_LICENCE, targetProvider: 'kafka');
+        $this->warmUpConsumer();
+        $this->fillOutbox();
+    }
+
+    public function setUpRelayBatchedIntoRedisTarget(): void
+    {
+        $this->messaging = $this->bootstrapOutbox(licenceKey: LicenceTesting::VALID_LICENCE, targetProvider: 'redis');
+        $this->warmUpConsumer();
+        $this->fillOutbox();
+    }
+
+    public function setUpRelayBatchedIntoSqsTarget(): void
+    {
+        $this->messaging = $this->bootstrapOutbox(licenceKey: LicenceTesting::VALID_LICENCE, targetProvider: 'sqs');
+        $this->warmUpConsumer();
+        $this->fillOutbox();
+    }
+
+    #[BeforeMethods('setUpRelayMessageByMessage')]
+    public function bench_dbal_outbox_drain_message_by_message(): void
+    {
+        $this->drainWholeOutbox();
+    }
+
+    #[BeforeMethods('setUpRelayBatched')]
+    public function bench_dbal_outbox_drain_batched(): void
+    {
+        $this->drainWholeOutbox();
+    }
+
+    #[BeforeMethods('setUpRelaySingleBatch')]
+    public function bench_dbal_outbox_drain_as_single_batch(): void
+    {
+        $this->drainWholeOutbox();
+    }
+
+    #[BeforeMethods('setUpRelayBatchedIntoDbalTarget')]
+    public function bench_dbal_outbox_drain_batched_into_high_throughput_dbal_target(): void
+    {
+        $this->drainWholeOutbox();
+    }
+
+    #[BeforeMethods('setUpRelayBatchedIntoAmqpTarget'), Iterations(3)]
+    public function bench_dbal_outbox_drain_batched_into_rabbitmq_target(): void
+    {
+        $this->drainWholeOutbox();
+    }
+
+    #[BeforeMethods('setUpRelayBatchedIntoKafkaTarget'), Iterations(3)]
+    public function bench_dbal_outbox_drain_batched_into_kafka_target(): void
+    {
+        $this->drainWholeOutbox();
+    }
+
+    #[BeforeMethods('setUpRelayBatchedIntoRedisTarget'), Iterations(3)]
+    public function bench_dbal_outbox_drain_batched_into_redis_target(): void
+    {
+        $this->drainWholeOutbox();
+    }
+
+    #[BeforeMethods('setUpRelayBatchedIntoSqsTarget'), Iterations(3)]
+    public function bench_dbal_outbox_drain_batched_into_sqs_target(): void
+    {
+        $this->drainWholeOutbox();
+    }
+
+    private function warmUpConsumer(): void
+    {
+        $context = (new DbalConnectionFactory(self::databaseDsn()))->createContext();
+        $context->createDataBaseTable();
+        $context->purgeQueue($context->createQueue('benchmark_outbox'));
+        $context->purgeQueue($context->createQueue('benchmark_target'));
+
+        $this->messaging->sendCommandWithRoutingKey('benchmark.relayOrder', self::MESSAGE_PAYLOAD);
+        $this->messaging->run('benchmark_outbox', ExecutionPollingMetadata::createWithFinishWhenNoMessages());
+    }
+
+    private function drainWholeOutbox(): void
+    {
+        $this->messaging->run('benchmark_outbox', ExecutionPollingMetadata::createWithFinishWhenNoMessages());
+    }
+
+    private static function databaseDsn(): string
+    {
+        return getenv('DATABASE_DSN') ?: 'pgsql://ecotone:secret@localhost:5432/ecotone';
+    }
+
+    private function fillOutbox(): void
+    {
+        for ($messageNumber = 0; $messageNumber < self::AMOUNT_OF_RELAYED_MESSAGES; $messageNumber++) {
+            $this->messaging->sendCommandWithRoutingKey('benchmark.relayOrder', self::MESSAGE_PAYLOAD);
+        }
+    }
+
+    private function bootstrapOutbox(?string $licenceKey, ?int $maxForwardingBatchSize = null, string $targetProvider = 'in_memory'): FlowTestSupport
+    {
+        $targetName = in_array($targetProvider, ['in_memory', 'dbal'], true) ? 'benchmark_target' : uniqid('benchmark_target_');
+        if ($licenceKey !== null) {
+            $relayChannel = OutboxForwardingMessageChannel::create('benchmark_relay_orders', 'benchmark_outbox', $targetName);
+            if ($maxForwardingBatchSize !== null) {
+                $relayChannel = $relayChannel->withMaxForwardingBatchSize($maxForwardingBatchSize);
+            }
+        } else {
+            $relayChannel = CombinedMessageChannel::create('benchmark_relay_orders', ['benchmark_outbox', $targetName]);
+        }
+        [$targetChannel, $targetServices, $targetPackage] = match ($targetProvider) {
+            'in_memory' => [SimpleMessageChannelBuilder::createQueueChannel($targetName), [], null],
+            'dbal' => [DbalBackedMessageChannelBuilder::create($targetName)->withHighThroughputPublishing(), [], null],
+            'amqp' => [
+                AmqpBackedMessageChannelBuilder::create($targetName)->withHighThroughputPublishing(),
+                [AmqpConnectionFactory::class => new AmqpConnectionFactory(['dsn' => getenv('RABBIT_HOST') ?: 'amqp://guest:guest@localhost:5672/%2f'])],
+                ModulePackageList::AMQP_PACKAGE,
+            ],
+            'kafka' => [
+                KafkaMessageChannelBuilder::create($targetName, topicName: $targetName, messageGroupId: $targetName)->withHighThroughputPublishing(),
+                [KafkaBrokerConfiguration::class => KafkaBrokerConfiguration::createWithDefaults([getenv('KAFKA_DSN') ?: 'localhost:9094'])],
+                ModulePackageList::KAFKA_PACKAGE,
+            ],
+            'redis' => [
+                RedisBackedMessageChannelBuilder::create($targetName)->withHighThroughputPublishing(),
+                [RedisConnectionFactory::class => new RedisConnectionFactory(getenv('REDIS_DSN') ?: 'redis://localhost:6379')],
+                ModulePackageList::REDIS_PACKAGE,
+            ],
+            'sqs' => [
+                SqsBackedMessageChannelBuilder::create($targetName)->withHighThroughputPublishing(),
+                [SqsConnectionFactory::class => new SqsConnectionFactory(getenv('SQS_DSN') ?: 'sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localhost:4566&version=latest')],
+                ModulePackageList::SQS_PACKAGE,
+            ],
+        };
+
+        $orderService = new class () {
+            #[Asynchronous('benchmark_relay_orders')]
+            #[CommandHandler('benchmark.relayOrder', endpointId: 'benchmarkRelayOrderEndpoint')]
+            public function handle(string $order): void
+            {
+            }
+        };
+
+        return EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            array_merge(
+                [
+                    DbalConnectionFactory::class => new DbalConnectionFactory(self::databaseDsn()),
+                    $orderService,
+                ],
+                $targetServices,
+            ),
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept(array_merge(
+                    [ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE],
+                    $targetPackage !== null ? [$targetPackage] : [],
+                )))
+                ->withExtensionObjects([
+                    $relayChannel,
+                    DbalBackedMessageChannelBuilder::create('benchmark_outbox')
+                        ->withReceiveTimeout(20),
+                    $targetChannel,
+                ]),
+            licenceKey: $licenceKey,
+        );
+    }
+}
diff --git a/packages/Amqp/src/AmqpBackedMessageChannelBuilder.php b/packages/Amqp/src/AmqpBackedMessageChannelBuilder.php
index 3aa4a822b..cfa770b15 100644
--- a/packages/Amqp/src/AmqpBackedMessageChannelBuilder.php
+++ b/packages/Amqp/src/AmqpBackedMessageChannelBuilder.php
@@ -72,7 +72,7 @@ public function withPublisherConfirms(bool $enabled): self
         return $this;
     }
 
-    public function withAsyncPublishing(bool $enabled = true, ?int $timeoutInMilliseconds = null): self
+    public function withHighThroughputPublishing(bool $enabled = true, ?int $timeoutInMilliseconds = null): self
     {
         $this->getAmqpOutboundChannelAdapter()->withAsyncPublishing($enabled, $timeoutInMilliseconds);
 
diff --git a/packages/Amqp/tests/Integration/AsyncPublishingTest.php b/packages/Amqp/tests/Integration/AsyncPublishingTest.php
index 799b4b32d..7b131ccb9 100644
--- a/packages/Amqp/tests/Integration/AsyncPublishingTest.php
+++ b/packages/Amqp/tests/Integration/AsyncPublishingTest.php
@@ -220,7 +220,7 @@ public function test_batch_published_over_amqp_lib_connection_is_delivered(): vo
                 ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::AMQP_PACKAGE]))
                 ->withExtensionObjects([
                     AmqpBackedMessageChannelBuilder::create('asyncOrdersChannel', queueName: $channelName)
-                        ->withAsyncPublishing(),
+                        ->withHighThroughputPublishing(),
                 ]),
             licenceKey: LicenceTesting::VALID_LICENCE,
         );
@@ -309,7 +309,7 @@ private function bootstrapEcotone(string $channelName, object $orderService, ?st
                 ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::AMQP_PACKAGE]))
                 ->withExtensionObjects([
                     AmqpBackedMessageChannelBuilder::create('asyncOrdersChannel', queueName: $channelName)
-                        ->withAsyncPublishing(),
+                        ->withHighThroughputPublishing(),
                 ]),
             licenceKey: $licenceKey,
         );
diff --git a/packages/Dbal/src/BatchForwarding/DbalBatchForwardingModule.php b/packages/Dbal/src/BatchForwarding/DbalBatchForwardingModule.php
new file mode 100644
index 000000000..6a6bfdd8c
--- /dev/null
+++ b/packages/Dbal/src/BatchForwarding/DbalBatchForwardingModule.php
@@ -0,0 +1,127 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Dbal\BatchForwarding;
+
+use Ecotone\AnnotationFinder\AnnotationFinder;
+use Ecotone\Dbal\DbalBackedMessageChannelBuilder;
+use Ecotone\Dbal\DbalReconnectableConnectionFactory;
+use Ecotone\Dbal\OutboxForwardingMessageChannel;
+use Ecotone\Enqueue\CachedConnectionFactory;
+use Ecotone\Messaging\Attribute\ModuleAnnotation;
+use Ecotone\Messaging\Attribute\WithoutDatabaseTransaction;
+use Ecotone\Messaging\Attribute\WithoutMessageCollector;
+use Ecotone\Messaging\Config\Annotation\AnnotationModule;
+use Ecotone\Messaging\Config\Annotation\ModuleConfiguration\ExtensionObjectResolver;
+use Ecotone\Messaging\Config\Annotation\ModuleConfiguration\NoExternalConfigurationModule;
+use Ecotone\Messaging\Config\Configuration;
+use Ecotone\Messaging\Config\Container\AttributeDefinition;
+use Ecotone\Messaging\Config\Container\Definition;
+use Ecotone\Messaging\Config\Container\Reference;
+use Ecotone\Messaging\Config\ModulePackageList;
+use Ecotone\Messaging\Config\ModuleReferenceSearchService;
+use Ecotone\Messaging\Endpoint\InboundChannelAdapter\InboundChannelAdapterBuilder;
+use Ecotone\Messaging\Handler\ChannelResolver;
+use Ecotone\Messaging\Handler\InterfaceToCallRegistry;
+use Ecotone\Messaging\Handler\Logger\LoggingGateway;
+use Ecotone\Messaging\NullableMessageChannel;
+use Ecotone\Messaging\Scheduling\EcotoneClockInterface;
+use Ecotone\Messaging\Support\Assert;
+
+#[ModuleAnnotation]
+/**
+ * licence Enterprise
+ */
+final class DbalBatchForwardingModule extends NoExternalConfigurationModule implements AnnotationModule
+{
+    public static function create(AnnotationFinder $annotationRegistrationService, InterfaceToCallRegistry $interfaceToCallRegistry): static
+    {
+        return new self();
+    }
+
+    public function prepare(Configuration $messagingConfiguration, array $extensionObjects, ModuleReferenceSearchService $moduleReferenceSearchService, InterfaceToCallRegistry $interfaceToCallRegistry): void
+    {
+        $channelBuilders = [];
+        foreach (ExtensionObjectResolver::resolve(DbalBackedMessageChannelBuilder::class, $extensionObjects) as $channelBuilder) {
+            $channelBuilders[$channelBuilder->getMessageChannelName()] = $channelBuilder;
+        }
+
+        $forwardingChannelsPerSource = [];
+        /** @var OutboxForwardingMessageChannel $outboxForwardingChannel */
+        foreach (ExtensionObjectResolver::resolve(OutboxForwardingMessageChannel::class, $extensionObjects) as $outboxForwardingChannel) {
+            $sourceChannelName = $outboxForwardingChannel->getSourceChannelName();
+            if (isset($forwardingChannelsPerSource[$sourceChannelName])) {
+                $alreadyRegistered = $forwardingChannelsPerSource[$sourceChannelName];
+                Assert::isTrue(
+                    $alreadyRegistered->getMaxForwardingBatchSize() === $outboxForwardingChannel->getMaxForwardingBatchSize()
+                    && $alreadyRegistered->getEndpointId() === $outboxForwardingChannel->getEndpointId()
+                    && $alreadyRegistered->getFinalFailureStrategy() === $outboxForwardingChannel->getFinalFailureStrategy(),
+                    "Outbox forwarding Message Channels sharing source `{$sourceChannelName}` must configure the same batch size, endpoint id and failure strategy.",
+                );
+
+                continue;
+            }
+            $forwardingChannelsPerSource[$sourceChannelName] = $outboxForwardingChannel;
+        }
+
+        $outboxesPerEndpointId = [];
+        foreach ($forwardingChannelsPerSource as $sourceChannelName => $outboxForwardingChannel) {
+            $channelBuilder = $channelBuilders[$sourceChannelName] ?? null;
+            if ($channelBuilder === null) {
+                continue;
+            }
+
+            $messagingConfiguration->registerBatchForwardingSourceChannel($sourceChannelName);
+            $outboxesPerEndpointId[$outboxForwardingChannel->getEndpointId()][] = $this->createOutboxPublisherDefinition($channelBuilder, $outboxForwardingChannel);
+        }
+
+        foreach ($outboxesPerEndpointId as $endpointId => $outboxPublishers) {
+            $batchPublisherReference = 'dbal.batch_forwarding.' . $endpointId;
+            $messagingConfiguration->registerServiceDefinition(
+                $batchPublisherReference,
+                new Definition(DbalBatchPublisher::class, [$outboxPublishers]),
+            );
+            $messagingConfiguration->registerConsumer(
+                InboundChannelAdapterBuilder::create(
+                    NullableMessageChannel::CHANNEL_NAME,
+                    $batchPublisherReference,
+                    $interfaceToCallRegistry->getFor(DbalBatchPublisher::class, 'publish'),
+                )
+                    ->withEndpointId((string) $endpointId)
+                    ->withContinuousPolling()
+                    ->withEndpointAnnotations([new AttributeDefinition(WithoutMessageCollector::class), new AttributeDefinition(WithoutDatabaseTransaction::class)]),
+            );
+        }
+    }
+
+    public function canHandle($extensionObject): bool
+    {
+        return $extensionObject instanceof OutboxForwardingMessageChannel
+            || $extensionObject instanceof DbalBackedMessageChannelBuilder;
+    }
+
+    public function getModulePackageName(): string
+    {
+        return ModulePackageList::DBAL_PACKAGE;
+    }
+
+    private function createOutboxPublisherDefinition(DbalBackedMessageChannelBuilder $channelBuilder, OutboxForwardingMessageChannel $outboxForwardingChannel): Definition
+    {
+        $inboundChannelAdapter = $channelBuilder->getInboundChannelAdapter();
+
+        return new Definition(DbalOutboxPublisher::class, [
+            new Definition(CachedConnectionFactory::class, [
+                new Definition(DbalReconnectableConnectionFactory::class, [
+                    new Reference($inboundChannelAdapter->getConnectionReferenceName()),
+                ]),
+            ], 'createFor'),
+            $channelBuilder->getMessageChannelName(),
+            new Reference(ChannelResolver::class),
+            new Reference(LoggingGateway::class),
+            new Reference(EcotoneClockInterface::class),
+            $outboxForwardingChannel->getMaxForwardingBatchSize(),
+            $outboxForwardingChannel->getFinalFailureStrategy() ?? $inboundChannelAdapter->getFinalFailureStrategy(),
+        ]);
+    }
+}
diff --git a/packages/Dbal/src/BatchForwarding/DbalBatchPublisher.php b/packages/Dbal/src/BatchForwarding/DbalBatchPublisher.php
new file mode 100644
index 000000000..c687616e8
--- /dev/null
+++ b/packages/Dbal/src/BatchForwarding/DbalBatchPublisher.php
@@ -0,0 +1,43 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Dbal\BatchForwarding;
+
+/**
+ * licence Enterprise
+ */
+final class DbalBatchPublisher
+{
+    private const IDLE_POLL_INTERVAL_IN_MILLISECONDS = 100;
+
+    private bool $previousPublishFoundNothing = false;
+
+    /**
+     * @param DbalOutboxPublisher[] $outboxPublishers
+     */
+    public function __construct(private array $outboxPublishers)
+    {
+    }
+
+    public function publish(): ?int
+    {
+        $publishedAmount = 0;
+        foreach ($this->outboxPublishers as $outboxPublisher) {
+            $publishedAmount += $outboxPublisher->publishBatch();
+        }
+
+        if ($publishedAmount > 0) {
+            $this->previousPublishFoundNothing = false;
+
+            return $publishedAmount;
+        }
+
+        if ($this->previousPublishFoundNothing) {
+            usleep(self::IDLE_POLL_INTERVAL_IN_MILLISECONDS * 1000);
+        }
+        $this->previousPublishFoundNothing = true;
+
+        return null;
+    }
+}
diff --git a/packages/Dbal/src/BatchForwarding/DbalOutboxPublisher.php b/packages/Dbal/src/BatchForwarding/DbalOutboxPublisher.php
new file mode 100644
index 000000000..aff89a02b
--- /dev/null
+++ b/packages/Dbal/src/BatchForwarding/DbalOutboxPublisher.php
@@ -0,0 +1,374 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Dbal\BatchForwarding;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
+use Ecotone\Enqueue\CachedConnectionFactory;
+use Ecotone\Messaging\BatchMessage;
+use Ecotone\Messaging\Channel\BatchSupportingMessageChannel;
+use Ecotone\Messaging\Channel\MessageChannelInterceptorAdapter;
+use Ecotone\Messaging\Endpoint\FinalFailureStrategy;
+use Ecotone\Messaging\Endpoint\PollingConsumer\ConnectionException;
+use Ecotone\Messaging\Handler\ChannelResolver;
+use Ecotone\Messaging\Handler\Logger\LoggingGateway;
+use Ecotone\Messaging\MessageChannel;
+use Ecotone\Messaging\MessageHeaders;
+use Ecotone\Messaging\Scheduling\EcotoneClockInterface;
+use Ecotone\Messaging\Support\MessageBuilder;
+use Enqueue\Dbal\DbalContext;
+use Enqueue\Dbal\DbalType;
+use Symfony\Component\Uid\Uuid;
+use Throwable;
+
+/**
+ * Relays outbox rows without deserializing them: body stays an opaque payload in its stored content type,
+ * headers are the already storeable scalars written by the producing side. Format conversion belongs to
+ * the target channel when its media type differs, and to the final consumer when the handler is invoked.
+ *
+ * licence Enterprise
+ */
+final class DbalOutboxPublisher
+{
+    private const CLAIM_REDELIVERY_SAFETY_WINDOW_IN_SECONDS = 1200;
+
+    public function __construct(
+        private CachedConnectionFactory $connectionFactory,
+        private string $queueName,
+        private ChannelResolver $channelResolver,
+        private LoggingGateway $logger,
+        private EcotoneClockInterface $clock,
+        private int $maxBatchSize,
+        private FinalFailureStrategy $finalFailureStrategy,
+    ) {
+    }
+
+    public function publishBatch(): int
+    {
+        /** @var DbalContext $context */
+        $context = $this->connectionFactory->createContext();
+        $connection = $context->getDbalConnection();
+
+        $connection->beginTransaction();
+        try {
+            $drainedRecords = $this->claimPendingRecords($context, $connection);
+            if ($drainedRecords === []) {
+                $this->markExpiredClaimsForRedelivery($context, $connection);
+                $drainedRecords = $this->claimPendingRecords($context, $connection);
+            }
+            if ($drainedRecords === []) {
+                $connection->commit();
+
+                return 0;
+            }
+
+            $deliveredRowIds = [];
+            $releasedRowIds = [];
+            foreach ($this->groupByTargetChannel($drainedRecords, $releasedRowIds) as $groupTargetChannelName => $groupedRecords) {
+                $this->forwardGroup((string) $groupTargetChannelName, $groupedRecords, $deliveredRowIds, $releasedRowIds);
+            }
+
+            $this->deleteRows($context, $connection, $deliveredRowIds);
+            $this->releaseRows($context, $connection, $releasedRowIds);
+            $connection->commit();
+
+            return count($deliveredRowIds);
+        } catch (Throwable $exception) {
+            $connection->rollBack();
+
+            throw $exception;
+        }
+    }
+
+    private function markExpiredClaimsForRedelivery(DbalContext $context, Connection $connection): void
+    {
+        $connection->createQueryBuilder()
+            ->update($context->getTableName())
+            ->set('delivery_id', ':deliveryId')
+            ->set('redelivered', ':redelivered')
+            ->andWhere('queue = :queue')
+            ->andWhere('redeliver_after < :now')
+            ->andWhere('delivery_id IS NOT NULL')
+            ->setParameter('queue', $this->queueName)
+            ->setParameter('now', $this->clock->now()->unixTime()->inSeconds(), DbalType::BIGINT)
+            ->setParameter('deliveryId', null, DbalType::GUID)
+            ->setParameter('redelivered', true, DbalType::BOOLEAN)
+            ->executeStatement();
+    }
+
+    /**
+     * @return array<int, array{rowId: string, targetChannelName: string|null, payload: string, headers: array<string, mixed>}>
+     */
+    private function claimPendingRecords(DbalContext $context, Connection $connection): array
+    {
+        $nowInSeconds = $this->clock->now()->unixTime()->inSeconds();
+
+        $claimedRows = $this->supportsLockedFetch($connection)
+            ? $this->fetchPendingRowsWithLock($context, $connection, $nowInSeconds)
+            : $this->fetchPendingRowsWithClaimMarker($context, $connection, $nowInSeconds);
+        if ($claimedRows === []) {
+            return [];
+        }
+
+        $expiredRowIds = [];
+        $records = [];
+        foreach ($claimedRows as $claimedRow) {
+            if (! ($claimedRow['redelivered'] || empty($claimedRow['time_to_live']) || $claimedRow['time_to_live'] > $nowInSeconds)) {
+                $expiredRowIds[] = $claimedRow['id'];
+
+                continue;
+            }
+
+            $records[] = $this->convertRowToForwardableRecord($claimedRow);
+        }
+        $this->deleteRows($context, $connection, $expiredRowIds);
+
+        return $records;
+    }
+
+    /**
+     * @param array<string, mixed> $claimedRow
+     * @return array{rowId: string, targetChannelName: string|null, payload: string, headers: array<string, mixed>}
+     */
+    private function convertRowToForwardableRecord(array $claimedRow): array
+    {
+        $headers = $claimedRow['properties'] ? json_decode((string) $claimedRow['properties'], true, 512, JSON_THROW_ON_ERROR) : [];
+        unset($headers[MessageHeaders::POLLED_CHANNEL_NAME], $headers[MessageHeaders::CONSUMER_POLLING_METADATA], $headers[MessageHeaders::CONSUMER_ACK_HEADER_LOCATION]);
+
+        $routingSlipChannels = array_filter(explode(',', (string) ($headers[MessageHeaders::ROUTING_SLIP] ?? '')));
+        $targetChannelName = array_shift($routingSlipChannels);
+        if ($routingSlipChannels === []) {
+            unset($headers[MessageHeaders::ROUTING_SLIP]);
+        } else {
+            $headers[MessageHeaders::ROUTING_SLIP] = implode(',', $routingSlipChannels);
+        }
+
+        return [
+            'rowId' => $claimedRow['id'],
+            'targetChannelName' => $targetChannelName,
+            'payload' => (string) $claimedRow['body'],
+            'headers' => $headers,
+        ];
+    }
+
+    /**
+     * @return array<int, array<string, mixed>>
+     */
+    private function fetchPendingRowsWithLock(DbalContext $context, Connection $connection, int $nowInSeconds): array
+    {
+        $lockedFetchSql = sprintf(
+            'SELECT * FROM %s WHERE queue = :queue AND (delayed_until IS NULL OR delayed_until <= :now) AND delivery_id IS NULL ORDER BY priority ASC, published_at ASC LIMIT %d FOR UPDATE SKIP LOCKED',
+            $context->getTableName(),
+            $this->maxBatchSize,
+        );
+
+        return $connection->executeQuery(
+            $lockedFetchSql,
+            ['queue' => $this->queueName, 'now' => $nowInSeconds],
+            ['now' => DbalType::INTEGER],
+        )->fetchAllAssociative();
+    }
+
+    /**
+     * @return array<int, array<string, mixed>>
+     */
+    private function fetchPendingRowsWithClaimMarker(DbalContext $context, Connection $connection, int $nowInSeconds): array
+    {
+        $selectedRows = $connection->createQueryBuilder()
+            ->select('*')
+            ->from($context->getTableName())
+            ->andWhere('queue = :queue')
+            ->andWhere('delayed_until IS NULL OR delayed_until <= :now')
+            ->andWhere('delivery_id IS NULL')
+            ->addOrderBy('priority', 'asc')
+            ->addOrderBy('published_at', 'asc')
+            ->setMaxResults($this->maxBatchSize)
+            ->setParameter('queue', $this->queueName)
+            ->setParameter('now', $nowInSeconds, DbalType::INTEGER)
+            ->executeQuery()
+            ->fetchAllAssociative();
+        if ($selectedRows === []) {
+            return [];
+        }
+
+        $batchDeliveryId = Uuid::v7()->toRfc4122();
+        $claimedAmount = $connection->createQueryBuilder()
+            ->update($context->getTableName())
+            ->set('delivery_id', ':deliveryId')
+            ->set('redeliver_after', ':redeliverAfter')
+            ->andWhere('id IN (:rowIds)')
+            ->andWhere('delivery_id IS NULL')
+            ->setParameter('deliveryId', $batchDeliveryId, DbalType::GUID)
+            ->setParameter('redeliverAfter', $nowInSeconds + self::CLAIM_REDELIVERY_SAFETY_WINDOW_IN_SECONDS, DbalType::BIGINT)
+            ->setParameter('rowIds', array_column($selectedRows, 'id'), $this->arrayOfStringsParameterType())
+            ->executeStatement();
+        if ($claimedAmount === 0) {
+            return [];
+        }
+        if ($claimedAmount === count($selectedRows)) {
+            return $selectedRows;
+        }
+
+        return $connection->createQueryBuilder()
+            ->select('*')
+            ->from($context->getTableName())
+            ->andWhere('delivery_id = :deliveryId')
+            ->addOrderBy('priority', 'asc')
+            ->addOrderBy('published_at', 'asc')
+            ->setParameter('deliveryId', $batchDeliveryId, DbalType::GUID)
+            ->executeQuery()
+            ->fetchAllAssociative();
+    }
+
+    private function supportsLockedFetch(Connection $connection): bool
+    {
+        return $connection->getDatabasePlatform() instanceof PostgreSQLPlatform;
+    }
+
+    /**
+     * @param array<int, array{rowId: string, targetChannelName: string|null, payload: string, headers: array<string, mixed>}> $drainedRecords
+     * @param string[] $releasedRowIds
+     * @return array<string, array<int, array{rowId: string, targetChannelName: string|null, payload: string, headers: array<string, mixed>}>>
+     */
+    private function groupByTargetChannel(array $drainedRecords, array &$releasedRowIds): array
+    {
+        $groups = [];
+        foreach ($drainedRecords as $drainedRecord) {
+            $targetChannelName = $drainedRecord['targetChannelName'];
+            if ($targetChannelName === null) {
+                $releasedRowIds[] = $drainedRecord['rowId'];
+                $this->logger->error(
+                    sprintf('Message with id `%s` inside outbox Channel `%s` has no routing slip to determine the forwarding target. It was released for redelivery.', $drainedRecord['headers'][MessageHeaders::MESSAGE_ID] ?? 'unknown', $this->queueName),
+                );
+
+                continue;
+            }
+            $groups[$targetChannelName][] = $drainedRecord;
+        }
+
+        return $groups;
+    }
+
+    /**
+     * @param array<int, array{rowId: string, targetChannelName: string|null, payload: string, headers: array<string, mixed>}> $groupedRecords
+     * @param string[] $deliveredRowIds
+     * @param string[] $releasedRowIds
+     */
+    private function forwardGroup(string $targetChannelName, array $groupedRecords, array &$deliveredRowIds, array &$releasedRowIds): void
+    {
+        $targetChannel = $this->channelResolver->resolve($targetChannelName);
+
+        if ($this->supportsBatchMessages($targetChannel)) {
+            try {
+                $targetChannel->send(
+                    MessageBuilder::withPayload(
+                        BatchMessage::fromEntries(array_map(
+                            fn (array $groupedRecord) => ['payload' => $groupedRecord['payload'], 'headers' => $groupedRecord['headers']],
+                            $groupedRecords,
+                        )),
+                    )->build(),
+                );
+            } catch (Throwable $exception) {
+                $this->handleFailedDelivery($groupedRecords, $targetChannelName, $exception, $deliveredRowIds, $releasedRowIds);
+
+                return;
+            }
+            foreach ($groupedRecords as $groupedRecord) {
+                $deliveredRowIds[] = $groupedRecord['rowId'];
+            }
+
+            return;
+        }
+
+        foreach ($groupedRecords as $groupedRecord) {
+            try {
+                $targetChannel->send(
+                    MessageBuilder::withPayload($groupedRecord['payload'])
+                        ->setMultipleHeaders($groupedRecord['headers'])
+                        ->build(),
+                );
+            } catch (Throwable $exception) {
+                $this->handleFailedDelivery([$groupedRecord], $targetChannelName, $exception, $deliveredRowIds, $releasedRowIds);
+
+                continue;
+            }
+            $deliveredRowIds[] = $groupedRecord['rowId'];
+        }
+    }
+
+    /**
+     * @param array<int, array{rowId: string, targetChannelName: string|null, payload: string, headers: array<string, mixed>}> $failedRecords
+     * @param string[] $deliveredRowIds
+     * @param string[] $releasedRowIds
+     */
+    private function handleFailedDelivery(array $failedRecords, string $targetChannelName, Throwable $exception, array &$deliveredRowIds, array &$releasedRowIds): void
+    {
+        if ($exception instanceof ConnectionException || $this->finalFailureStrategy === FinalFailureStrategy::STOP) {
+            throw $exception;
+        }
+
+        foreach ($failedRecords as $failedRecord) {
+            if ($this->finalFailureStrategy === FinalFailureStrategy::IGNORE) {
+                $deliveredRowIds[] = $failedRecord['rowId'];
+            } else {
+                $releasedRowIds[] = $failedRecord['rowId'];
+            }
+            $this->logger->info(
+                sprintf('Message with id `%s` handled with `%s` failure strategy, as delivery to `%s` failed. Due to %s', $failedRecord['headers'][MessageHeaders::MESSAGE_ID] ?? 'unknown', $this->finalFailureStrategy->value, $targetChannelName, $exception->getMessage()),
+                ['exception' => $exception],
+            );
+        }
+    }
+
+    /**
+     * @param string[] $rowIds
+     */
+    private function deleteRows(DbalContext $context, Connection $connection, array $rowIds): void
+    {
+        if ($rowIds === []) {
+            return;
+        }
+
+        $connection->createQueryBuilder()
+            ->delete($context->getTableName())
+            ->andWhere('id IN (:rowIds)')
+            ->setParameter('rowIds', $rowIds, $this->arrayOfStringsParameterType())
+            ->executeStatement();
+    }
+
+    /**
+     * @param string[] $rowIds
+     */
+    private function releaseRows(DbalContext $context, Connection $connection, array $rowIds): void
+    {
+        if ($rowIds === []) {
+            return;
+        }
+
+        $connection->createQueryBuilder()
+            ->update($context->getTableName())
+            ->set('delivery_id', ':deliveryId')
+            ->set('redelivered', ':redelivered')
+            ->andWhere('id IN (:rowIds)')
+            ->setParameter('deliveryId', null, DbalType::GUID)
+            ->setParameter('redelivered', true, DbalType::BOOLEAN)
+            ->setParameter('rowIds', $rowIds, $this->arrayOfStringsParameterType())
+            ->executeStatement();
+    }
+
+    private function supportsBatchMessages(MessageChannel $channel): bool
+    {
+        $unwrappedChannel = $channel instanceof MessageChannelInterceptorAdapter ? $channel->getInternalMessageChannel() : $channel;
+
+        return $unwrappedChannel instanceof BatchSupportingMessageChannel && $unwrappedChannel->supportsBatchMessages();
+    }
+
+    private function arrayOfStringsParameterType(): mixed
+    {
+        return class_exists('\Doctrine\DBAL\ArrayParameterType')
+            ? \Doctrine\DBAL\ArrayParameterType::STRING
+            : (defined('\Doctrine\DBAL\Connection::PARAM_STR_ARRAY') ? Connection::PARAM_STR_ARRAY : 'string[]');
+    }
+}
diff --git a/packages/Dbal/src/DbalBackedMessageChannelBuilder.php b/packages/Dbal/src/DbalBackedMessageChannelBuilder.php
index 29adc9994..e08f96825 100644
--- a/packages/Dbal/src/DbalBackedMessageChannelBuilder.php
+++ b/packages/Dbal/src/DbalBackedMessageChannelBuilder.php
@@ -31,9 +31,9 @@ public static function create(string $channelName, string $connectionReferenceNa
         return new self($channelName, $connectionReferenceName);
     }
 
-    public function withAsyncPublishing(bool $asyncPublishing = true): self
+    public function withHighThroughputPublishing(bool $enabled = true): self
     {
-        $this->getDbalOutboundChannelAdapter()->withAsyncPublishing($asyncPublishing);
+        $this->getDbalOutboundChannelAdapter()->withAsyncPublishing($enabled);
 
         return $this;
     }
diff --git a/packages/Dbal/src/DbalInboundChannelAdapterBuilder.php b/packages/Dbal/src/DbalInboundChannelAdapterBuilder.php
index 91a5b19b2..26471a8f3 100644
--- a/packages/Dbal/src/DbalInboundChannelAdapterBuilder.php
+++ b/packages/Dbal/src/DbalInboundChannelAdapterBuilder.php
@@ -38,6 +38,7 @@ public function compile(MessagingContainerBuilder $builder): Definition
             DefaultHeaderMapper::createWith($this->headerMapper, []),
             EnqueueHeader::HEADER_ACKNOWLEDGE,
             Reference::to(LoggingGateway::class),
+            $this->finalFailureStrategy,
         ]);
 
         return new Definition(DbalInboundChannelAdapter::class, [
diff --git a/packages/Dbal/src/DbalOutboundChannelAdapter.php b/packages/Dbal/src/DbalOutboundChannelAdapter.php
index 20cf2f273..7dc72077a 100644
--- a/packages/Dbal/src/DbalOutboundChannelAdapter.php
+++ b/packages/Dbal/src/DbalOutboundChannelAdapter.php
@@ -70,7 +70,7 @@ protected function handleBatch(BatchMessage $batchMessage, Context $context): vo
     {
         $messagesToSend = [];
         foreach ($batchMessage->getEntries() as $entry) {
-            $outboundMessage = $this->prepareOutboundMessage($this->convertBatchEntryToMessage($entry));
+            $outboundMessage = $this->prepareOutboundMessageFromBatchEntry($entry);
             $headers = $outboundMessage->getHeaders();
             $headers[MessageHeaders::CONTENT_TYPE] = $outboundMessage->getContentType();
 
diff --git a/packages/Dbal/src/Deduplication/DeduplicationInterceptor.php b/packages/Dbal/src/Deduplication/DeduplicationInterceptor.php
index a12dbfd71..8d6a7108e 100644
--- a/packages/Dbal/src/Deduplication/DeduplicationInterceptor.php
+++ b/packages/Dbal/src/Deduplication/DeduplicationInterceptor.php
@@ -11,6 +11,7 @@
 use Ecotone\Messaging\Attribute\AsynchronousRunningEndpoint;
 use Ecotone\Messaging\Attribute\Deduplicated;
 use Ecotone\Messaging\Attribute\IdentifiedAnnotation;
+use Ecotone\Messaging\Attribute\WithoutDatabaseTransaction;
 use Ecotone\Messaging\Handler\ClosureExpression\AttributeExpressionExecutor;
 use Ecotone\Messaging\Handler\ClosureExpression\ExecutorFor;
 use Ecotone\Messaging\Handler\Logger\LoggingGateway;
@@ -50,8 +51,12 @@ public function __construct(
         }
     }
 
-    public function deduplicate(MethodInvocation $methodInvocation, Message $message, #[ExecutorFor(Deduplicated::class)] ?AttributeExpressionExecutor $deduplicated, ?IdentifiedAnnotation $identifiedAnnotation, ?AsynchronousRunningEndpoint $asynchronousRunningEndpoint): mixed
+    public function deduplicate(MethodInvocation $methodInvocation, Message $message, #[ExecutorFor(Deduplicated::class)] ?AttributeExpressionExecutor $deduplicated, ?IdentifiedAnnotation $identifiedAnnotation, ?AsynchronousRunningEndpoint $asynchronousRunningEndpoint, ?WithoutDatabaseTransaction $withoutDatabaseTransaction = null): mixed
     {
+        if ($withoutDatabaseTransaction !== null) {
+            return $methodInvocation->proceed();
+        }
+
         $connectionFactory = CachedConnectionFactory::createFor(new DbalReconnectableConnectionFactory($this->connection));
         $contextId = spl_object_id($connectionFactory->createContext());
 
diff --git a/packages/Dbal/src/EnqueueDbal/DbalProducer.php b/packages/Dbal/src/EnqueueDbal/DbalProducer.php
index ad3121d48..ed803255d 100644
--- a/packages/Dbal/src/EnqueueDbal/DbalProducer.php
+++ b/packages/Dbal/src/EnqueueDbal/DbalProducer.php
@@ -4,6 +4,7 @@
 
 namespace Enqueue\Dbal;
 
+use Doctrine\DBAL\ParameterType;
 use Ecotone\Messaging\Scheduling\Duration;
 use Interop\Queue\Destination;
 use Interop\Queue\Exception\Exception;
@@ -22,21 +23,21 @@ class DbalProducer implements Producer
 {
     private const BATCH_INSERT_CHUNK_SIZE = 250;
 
-    private const COLUMN_TYPES = [
-        'id' => DbalType::GUID,
-        'published_at' => DbalType::INTEGER,
-        'body' => DbalType::TEXT,
-        'headers' => DbalType::TEXT,
-        'properties' => DbalType::TEXT,
-        'priority' => DbalType::SMALLINT,
-        'queue' => DbalType::STRING,
-        'redelivered' => DbalType::SMALLINT,
-        'delivery_id' => DbalType::STRING,
-        'redeliver_after' => DbalType::BIGINT,
-        'delayed_until' => DbalType::INTEGER,
-        'time_to_live' => DbalType::INTEGER,
+    private const PARAMETERIZED_COLUMN_BINDINGS = [
+        'id' => ParameterType::STRING,
+        'published_at' => ParameterType::INTEGER,
+        'body' => ParameterType::STRING,
+        'headers' => ParameterType::STRING,
+        'properties' => ParameterType::STRING,
+        'priority' => ParameterType::INTEGER,
+        'queue' => ParameterType::STRING,
+        'delayed_until' => ParameterType::INTEGER,
+        'time_to_live' => ParameterType::INTEGER,
     ];
 
+    private const CONSTANT_COLUMNS = 'redelivered, delivery_id, redeliver_after';
+    private const CONSTANT_COLUMN_VALUES = 'FALSE, NULL, NULL';
+
     /**
      * @var int|null
      */
@@ -105,22 +106,23 @@ public function sendBatch(Destination $destination, array $messages): void
 
     private function insertRecords(array $records): int
     {
-        $columns = array_keys(self::COLUMN_TYPES);
-        $rowPlaceholders = '(' . implode(', ', array_fill(0, count($columns), '?')) . ')';
+        $parameterizedColumns = array_keys(self::PARAMETERIZED_COLUMN_BINDINGS);
+        $rowPlaceholders = '(' . implode(', ', array_fill(0, count($parameterizedColumns), '?')) . ', ' . self::CONSTANT_COLUMN_VALUES . ')';
 
         $sql = sprintf(
-            'INSERT INTO %s (%s) VALUES %s',
+            'INSERT INTO %s (%s, %s) VALUES %s',
             $this->context->getTableName(),
-            implode(', ', $columns),
+            implode(', ', $parameterizedColumns),
+            self::CONSTANT_COLUMNS,
             implode(', ', array_fill(0, count($records), $rowPlaceholders)),
         );
 
         $parameters = [];
         $types = [];
         foreach ($records as $record) {
-            foreach ($columns as $column) {
+            foreach (self::PARAMETERIZED_COLUMN_BINDINGS as $column => $bindingType) {
                 $parameters[] = $record[$column];
-                $types[] = self::COLUMN_TYPES[$column];
+                $types[] = $record[$column] === null ? ParameterType::NULL : $bindingType;
             }
         }
 
diff --git a/packages/Dbal/src/ObjectManager/ObjectManagerInterceptor.php b/packages/Dbal/src/ObjectManager/ObjectManagerInterceptor.php
index 4d991f6e6..68b4191d9 100644
--- a/packages/Dbal/src/ObjectManager/ObjectManagerInterceptor.php
+++ b/packages/Dbal/src/ObjectManager/ObjectManagerInterceptor.php
@@ -7,6 +7,7 @@
 use Ecotone\Dbal\EcotoneManagerRegistryConnectionFactory;
 use Ecotone\Dbal\MultiTenant\MultiTenantConnectionFactory;
 use Ecotone\Messaging\Attribute\Parameter\Reference;
+use Ecotone\Messaging\Attribute\WithoutDatabaseTransaction;
 use Ecotone\Messaging\Handler\Logger\LoggingGateway;
 use Ecotone\Messaging\Handler\Processor\MethodInvoker\MethodInvocation;
 use Ecotone\Messaging\Message;
@@ -27,8 +28,12 @@ public function __construct(private array $managerRegistryConnectionFactories)
     {
     }
 
-    public function transactional(MethodInvocation $methodInvocation, Message $message, #[Reference] LoggingGateway $logger)
+    public function transactional(MethodInvocation $methodInvocation, Message $message, #[Reference] LoggingGateway $logger, ?WithoutDatabaseTransaction $withoutDatabaseTransaction = null)
     {
+        if ($withoutDatabaseTransaction !== null) {
+            return $methodInvocation->proceed();
+        }
+
         /** @var ManagerRegistry[] $managerRegistries */
         $managerRegistries = [];
 
diff --git a/packages/Dbal/src/OutboxForwardingMessageChannel.php b/packages/Dbal/src/OutboxForwardingMessageChannel.php
new file mode 100644
index 000000000..d035f1ba8
--- /dev/null
+++ b/packages/Dbal/src/OutboxForwardingMessageChannel.php
@@ -0,0 +1,101 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Dbal;
+
+use Ecotone\Messaging\Channel\CombinedMessageChannel;
+use Ecotone\Messaging\Channel\MessageChannelBuilder;
+use Ecotone\Messaging\Channel\OutboxForwardingChannel;
+use Ecotone\Messaging\Endpoint\FinalFailureStrategy;
+use Ecotone\Messaging\Support\Assert;
+
+/**
+ * Combined Message Channel made of exactly one Dbal backed outbox source and one target channel,
+ * published in batches by a dedicated forwarding endpoint instead of a message channel consumer.
+ * The source can be given as the Dbal channel builder itself, registering the outbox channel along the way.
+ *
+ * licence Enterprise
+ */
+final class OutboxForwardingMessageChannel extends CombinedMessageChannel implements OutboxForwardingChannel
+{
+    public const DEFAULT_MAX_FORWARDING_BATCH_SIZE = 100;
+
+    private int $maxForwardingBatchSize = self::DEFAULT_MAX_FORWARDING_BATCH_SIZE;
+    private ?string $endpointId = null;
+    private ?FinalFailureStrategy $finalFailureStrategy = null;
+    private ?DbalBackedMessageChannelBuilder $embeddedSourceChannelBuilder = null;
+
+    public static function create(string $referenceName, array|string|DbalBackedMessageChannelBuilder $sourceChannelName = [], ?string $targetChannelName = null): static
+    {
+        $embeddedSourceChannelBuilder = null;
+        if ($sourceChannelName instanceof DbalBackedMessageChannelBuilder) {
+            $embeddedSourceChannelBuilder = $sourceChannelName;
+            $sourceChannelName = $embeddedSourceChannelBuilder->getMessageChannelName();
+        }
+
+        $combinedChannels = is_string($sourceChannelName) ? [$sourceChannelName, $targetChannelName] : $sourceChannelName;
+        Assert::isTrue(count($combinedChannels) === 2, "Outbox forwarding Message Channel `{$referenceName}` requires exactly one source outbox channel and one target channel.");
+        [$outboxChannelName, $forwardingTargetChannelName] = $combinedChannels;
+        Assert::isTrue(is_string($outboxChannelName) && is_string($forwardingTargetChannelName), "Outbox forwarding Message Channel `{$referenceName}` requires channel names to be strings.");
+        Assert::isTrue($outboxChannelName !== $forwardingTargetChannelName, "Outbox forwarding Message Channel `{$referenceName}` requires source and target to be different channels.");
+
+        $outboxForwardingChannel = new static($referenceName, $combinedChannels);
+        $outboxForwardingChannel->embeddedSourceChannelBuilder = $embeddedSourceChannelBuilder;
+
+        return $outboxForwardingChannel;
+    }
+
+    public function withMaxForwardingBatchSize(int $maxForwardingBatchSize): self
+    {
+        Assert::isTrue($maxForwardingBatchSize > 0, 'Max forwarding batch size must be a positive number.');
+        $this->maxForwardingBatchSize = $maxForwardingBatchSize;
+
+        return $this;
+    }
+
+    public function withEndpointId(string $endpointId): self
+    {
+        Assert::notNullAndEmpty($endpointId, 'Endpoint id for outbox forwarding can not be empty.');
+        $this->endpointId = $endpointId;
+
+        return $this;
+    }
+
+    public function withFinalFailureStrategy(FinalFailureStrategy $finalFailureStrategy): self
+    {
+        $this->finalFailureStrategy = $finalFailureStrategy;
+
+        return $this;
+    }
+
+    public function getSourceChannelName(): string
+    {
+        return $this->getCombinedChannels()[0];
+    }
+
+    public function getTargetChannelName(): string
+    {
+        return $this->getCombinedChannels()[1];
+    }
+
+    public function getEmbeddedSourceChannelBuilder(): ?MessageChannelBuilder
+    {
+        return $this->embeddedSourceChannelBuilder;
+    }
+
+    public function getEndpointId(): string
+    {
+        return $this->endpointId ?? $this->getSourceChannelName();
+    }
+
+    public function getMaxForwardingBatchSize(): int
+    {
+        return $this->maxForwardingBatchSize;
+    }
+
+    public function getFinalFailureStrategy(): ?FinalFailureStrategy
+    {
+        return $this->finalFailureStrategy;
+    }
+}
diff --git a/packages/Dbal/tests/Fixture/BatchForwarding/AlwaysFailOnPayloadChannelInterceptor.php b/packages/Dbal/tests/Fixture/BatchForwarding/AlwaysFailOnPayloadChannelInterceptor.php
new file mode 100644
index 000000000..a6c9914f2
--- /dev/null
+++ b/packages/Dbal/tests/Fixture/BatchForwarding/AlwaysFailOnPayloadChannelInterceptor.php
@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Fixture\BatchForwarding;
+
+use Ecotone\Messaging\Channel\AbstractChannelInterceptor;
+use Ecotone\Messaging\Message;
+use Ecotone\Messaging\MessageChannel;
+use RuntimeException;
+
+/**
+ * licence Apache-2.0
+ */
+final class AlwaysFailOnPayloadChannelInterceptor extends AbstractChannelInterceptor
+{
+    public function __construct(private string $failingPayload)
+    {
+    }
+
+    public function preSend(Message $message, MessageChannel $messageChannel): ?Message
+    {
+        if ($message->getPayload() === $this->failingPayload) {
+            throw new RuntimeException('Delivery of ' . $this->failingPayload . ' failed');
+        }
+
+        return $message;
+    }
+}
diff --git a/packages/Dbal/tests/Fixture/BatchForwarding/FailOnceOnPayloadChannelInterceptor.php b/packages/Dbal/tests/Fixture/BatchForwarding/FailOnceOnPayloadChannelInterceptor.php
new file mode 100644
index 000000000..6e57c9583
--- /dev/null
+++ b/packages/Dbal/tests/Fixture/BatchForwarding/FailOnceOnPayloadChannelInterceptor.php
@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Fixture\BatchForwarding;
+
+use Ecotone\Messaging\Channel\AbstractChannelInterceptor;
+use Ecotone\Messaging\Message;
+use Ecotone\Messaging\MessageChannel;
+
+/**
+ * licence Apache-2.0
+ */
+final class FailOnceOnPayloadChannelInterceptor extends AbstractChannelInterceptor
+{
+    private bool $alreadyFailed = false;
+
+    public function __construct(private string $failingPayload, private string $exceptionClass)
+    {
+    }
+
+    public function preSend(Message $message, MessageChannel $messageChannel): ?Message
+    {
+        if (! $this->alreadyFailed && $message->getPayload() === $this->failingPayload) {
+            $this->alreadyFailed = true;
+
+            throw new ($this->exceptionClass)('Delivery of ' . $this->failingPayload . ' failed');
+        }
+
+        return $message;
+    }
+}
diff --git a/packages/Dbal/tests/Fixture/BatchForwarding/FailOnceOnPayloadPollableChannel.php b/packages/Dbal/tests/Fixture/BatchForwarding/FailOnceOnPayloadPollableChannel.php
new file mode 100644
index 000000000..671ce1f18
--- /dev/null
+++ b/packages/Dbal/tests/Fixture/BatchForwarding/FailOnceOnPayloadPollableChannel.php
@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Fixture\BatchForwarding;
+
+use Ecotone\Messaging\Channel\QueueChannel;
+use Ecotone\Messaging\Config\Container\Definition;
+use Ecotone\Messaging\Message;
+use RuntimeException;
+
+/**
+ * licence Apache-2.0
+ */
+final class FailOnceOnPayloadPollableChannel extends QueueChannel
+{
+    public function __construct(private string $failingPayload, private int $failuresLeft = 2)
+    {
+        parent::__construct('failingProcessing');
+    }
+
+    public function send(Message $message): void
+    {
+        if ($this->failuresLeft > 0 && $message->getPayload() === $this->failingPayload) {
+            $this->failuresLeft--;
+
+            throw new RuntimeException('Delivery of ' . $this->failingPayload . ' failed');
+        }
+
+        parent::send($message);
+    }
+
+    public function getDefinition(): Definition
+    {
+        return new Definition(self::class, [$this->failingPayload, $this->failuresLeft]);
+    }
+}
diff --git a/packages/Dbal/tests/Fixture/BatchForwarding/FailingPollableChannel.php b/packages/Dbal/tests/Fixture/BatchForwarding/FailingPollableChannel.php
new file mode 100644
index 000000000..acbdc1bfd
--- /dev/null
+++ b/packages/Dbal/tests/Fixture/BatchForwarding/FailingPollableChannel.php
@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Fixture\BatchForwarding;
+
+use Ecotone\Messaging\Channel\QueueChannel;
+use Ecotone\Messaging\Config\Container\Definition;
+use Ecotone\Messaging\Message;
+use RuntimeException;
+
+/**
+ * licence Apache-2.0
+ */
+final class FailingPollableChannel extends QueueChannel
+{
+    public function __construct()
+    {
+        parent::__construct('failingProcessing');
+    }
+
+    public function send(Message $message): void
+    {
+        throw new RuntimeException('Target channel is unavailable');
+    }
+
+    public function getDefinition(): Definition
+    {
+        return new Definition(self::class);
+    }
+}
diff --git a/packages/Dbal/tests/Integration/AsyncPublishingTest.php b/packages/Dbal/tests/Integration/AsyncPublishingTest.php
index ad13337cd..bdb4b0fbf 100644
--- a/packages/Dbal/tests/Integration/AsyncPublishingTest.php
+++ b/packages/Dbal/tests/Integration/AsyncPublishingTest.php
@@ -286,7 +286,7 @@ private function bootstrapEcotoneWithChannel(object $orderService, ?string $lice
                 ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
                 ->withExtensionObjects([
                     DbalBackedMessageChannelBuilder::create('asyncOrdersChannel')
-                        ->withAsyncPublishing(),
+                        ->withHighThroughputPublishing(),
                 ]),
             licenceKey: $licenceKey,
         );
diff --git a/packages/Dbal/tests/Integration/CombinedChannelBatchForwardingTest.php b/packages/Dbal/tests/Integration/CombinedChannelBatchForwardingTest.php
new file mode 100644
index 000000000..d9a625542
--- /dev/null
+++ b/packages/Dbal/tests/Integration/CombinedChannelBatchForwardingTest.php
@@ -0,0 +1,1089 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Integration;
+
+use Ecotone\Dbal\DbalBackedMessageChannelBuilder;
+use Ecotone\Dbal\OutboxForwardingMessageChannel;
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Lite\Test\FlowTestSupport;
+use Ecotone\Messaging\Attribute\Asynchronous;
+use Ecotone\Messaging\Channel\CombinedMessageChannel;
+use Ecotone\Messaging\Channel\PollableChannel\GlobalPollableChannelConfiguration;
+use Ecotone\Messaging\Channel\PollableChannel\PollableChannelConfiguration;
+use Ecotone\Messaging\Channel\SimpleChannelInterceptorBuilder;
+use Ecotone\Messaging\Channel\SimpleMessageChannelBuilder;
+use Ecotone\Messaging\Config\ConfigurationException;
+use Ecotone\Messaging\Config\ModulePackageList;
+use Ecotone\Messaging\Config\ServiceConfiguration;
+use Ecotone\Messaging\Endpoint\ExecutionPollingMetadata;
+use Ecotone\Messaging\Endpoint\FinalFailureStrategy;
+use Ecotone\Messaging\Endpoint\PollingConsumer\ConnectionException;
+use Ecotone\Messaging\Handler\Recoverability\RetryTemplateBuilder;
+use Ecotone\Messaging\PollableChannel;
+use Ecotone\Messaging\Support\LicensingException;
+use Ecotone\Modelling\Attribute\CommandHandler;
+use Ecotone\Modelling\Attribute\QueryHandler;
+use Ecotone\Test\LicenceTesting;
+use Enqueue\Dbal\DbalConnectionFactory;
+use RuntimeException;
+use Test\Ecotone\Dbal\DbalMessagingTestCase;
+use Test\Ecotone\Dbal\Fixture\BatchForwarding\AlwaysFailOnPayloadChannelInterceptor;
+use Test\Ecotone\Dbal\Fixture\BatchForwarding\FailingPollableChannel;
+use Test\Ecotone\Dbal\Fixture\BatchForwarding\FailOnceOnPayloadChannelInterceptor;
+use Test\Ecotone\Dbal\Fixture\BatchForwarding\FailOnceOnPayloadPollableChannel;
+
+/**
+ * licence Apache-2.0
+ * @internal
+ */
+final class CombinedChannelBatchForwardingTest extends DbalMessagingTestCase
+{
+    public function test_single_run_of_outbox_consumer_moves_multiple_messages_to_target_asynchronous_channel(): void
+    {
+        $orderService = new class () {
+            /** @var string[] */
+            private array $orders = [];
+
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+                $this->orders[] = $order;
+            }
+
+            #[QueryHandler('order.getRegistered')]
+            public function getRegistered(): array
+            {
+                return $this->orders;
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', DbalBackedMessageChannelBuilder::create('outbox'), 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso', 'latte', 'cappuccino'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+        $this->assertSame(0, $this->amountOfRowsOn('outbox'));
+    }
+
+    public function test_forwarding_as_single_batch_to_target_with_high_throughput_publishing_delivers_all_messages(): void
+    {
+        $orderService = new class () {
+            /** @var string[] */
+            private array $orders = [];
+
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+                $this->orders[] = $order;
+            }
+
+            #[QueryHandler('order.getRegistered')]
+            public function getRegistered(): array
+            {
+                return $this->orders;
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing')
+                        ->withHighThroughputPublishing(),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+        $this->assertNull($messaging->getMessageChannel('outbox')->receive());
+
+        $messaging->run('orderProcessing', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 3, maxExecutionTimeInMilliseconds: 10000));
+
+        $this->assertSame(['espresso', 'latte', 'cappuccino'], $messaging->sendQueryWithRouting('order.getRegistered'));
+    }
+
+    public function test_single_run_moves_no_more_messages_than_configured_forwarding_batch_size(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing')
+                        ->withMaxForwardingBatchSize(2),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso', 'latte'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+        $this->assertSame(['cappuccino'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('outbox'))));
+    }
+
+    public function test_execution_limit_of_two_moves_exactly_two_batches(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing')
+                        ->withMaxForwardingBatchSize(100),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        for ($messageNumber = 0; $messageNumber < 300; $messageNumber++) {
+            $messaging->sendCommandWithRoutingKey('order.register', 'order-' . $messageNumber);
+        }
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, maxExecutionTimeInMilliseconds: 30000));
+
+        $expectedMoved = array_map(fn (int $messageNumber) => 'order-' . $messageNumber, range(0, 199));
+        $expectedRemaining = array_map(fn (int $messageNumber) => 'order-' . $messageNumber, range(200, 299));
+        $this->assertSame($expectedMoved, $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+        $this->assertSame($expectedRemaining, $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('outbox'))));
+    }
+
+    public function test_plain_combined_channel_reusing_outbox_forwarding_source_fails_at_compile_time(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+
+            #[Asynchronous('otherOrders')]
+            #[CommandHandler('order.registerOther', endpointId: 'orderRegisterOtherEndpoint')]
+            public function registerOther(string $order): void
+            {
+            }
+        };
+
+        $this->expectException(ConfigurationException::class);
+
+        EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    CombinedMessageChannel::create('otherOrders', ['outbox', 'otherProcessing']),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('otherProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+    }
+
+    public function test_batch_forwarding_configuration_requires_enterprise_licence(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $this->expectException(LicensingException::class);
+
+        EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+        );
+    }
+
+    public function test_combined_channel_without_licence_keeps_one_message_per_run(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    CombinedMessageChannel::create('orders', ['outbox', 'orderProcessing']),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertCount(1, $this->receiveAllFrom($messaging->getMessageChannel('orderProcessing')));
+        $this->assertCount(2, $this->receiveAllFrom($messaging->getMessageChannel('outbox')));
+    }
+
+    public function test_messages_for_different_targets_on_shared_outbox_reach_their_own_channels(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('standardOrders')]
+            #[CommandHandler('order.registerStandard', endpointId: 'standardOrderEndpoint')]
+            public function registerStandard(string $order): void
+            {
+            }
+
+            #[Asynchronous('priorityOrders')]
+            #[CommandHandler('order.registerPriority', endpointId: 'priorityOrderEndpoint')]
+            public function registerPriority(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('standardOrders', 'outbox', 'standardProcessing')
+                        ->withMaxForwardingBatchSize(2),
+                    OutboxForwardingMessageChannel::create('priorityOrders', 'outbox', 'priorityProcessing')
+                        ->withMaxForwardingBatchSize(2),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('standardProcessing'),
+                    DbalBackedMessageChannelBuilder::create('priorityProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.registerStandard', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.registerStandard', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.registerPriority', 'flat white');
+        $messaging->sendCommandWithRoutingKey('order.registerPriority', 'cortado');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 1, maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso', 'latte'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('standardProcessing'))));
+        $this->assertSame(0, $this->amountOfRowsOn('priorityProcessing'));
+        $this->assertSame(2, $this->amountOfRowsOn('outbox'));
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 1, maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['flat white', 'cortado'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('priorityProcessing'))));
+        $this->assertSame(0, $this->amountOfPendingRowsOn('standardProcessing'));
+        $this->assertSame(0, $this->amountOfRowsOn('outbox'));
+    }
+
+    public function test_plain_asynchronous_dbal_channel_keeps_one_message_per_handled_message(): void
+    {
+        $orderService = new class () {
+            /** @var string[] */
+            private array $orders = [];
+
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+                $this->orders[] = $order;
+            }
+
+            #[QueryHandler('order.getRegistered')]
+            public function getRegistered(): array
+            {
+                return $this->orders;
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    DbalBackedMessageChannelBuilder::create('orders'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+
+        $messaging->run('orders', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso'], $messaging->sendQueryWithRouting('order.getRegistered'));
+    }
+
+    public function test_multiple_outbox_channels_are_published_by_single_shared_endpoint(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('standardOrders')]
+            #[CommandHandler('order.registerStandard', endpointId: 'standardOrderEndpoint')]
+            public function registerStandard(string $order): void
+            {
+            }
+
+            #[Asynchronous('priorityOrders')]
+            #[CommandHandler('order.registerPriority', endpointId: 'priorityOrderEndpoint')]
+            public function registerPriority(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('standardOrders', 'standardOutbox', 'standardProcessing')
+                        ->withEndpointId('sharedOutboxPublisher'),
+                    OutboxForwardingMessageChannel::create('priorityOrders', 'priorityOutbox', 'priorityProcessing')
+                        ->withEndpointId('sharedOutboxPublisher'),
+                    DbalBackedMessageChannelBuilder::create('standardOutbox'),
+                    DbalBackedMessageChannelBuilder::create('priorityOutbox'),
+                    DbalBackedMessageChannelBuilder::create('standardProcessing'),
+                    DbalBackedMessageChannelBuilder::create('priorityProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.registerStandard', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.registerPriority', 'flat white');
+        $messaging->sendCommandWithRoutingKey('order.registerStandard', 'latte');
+
+        $messaging->run('sharedOutboxPublisher', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso', 'latte'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('standardProcessing'))));
+        $this->assertSame(['flat white'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('priorityProcessing'))));
+        $this->assertSame(0, $this->amountOfRowsOn('standardOutbox'));
+        $this->assertSame(0, $this->amountOfRowsOn('priorityOutbox'));
+    }
+
+    public function test_failed_send_of_single_message_on_target_channel_releases_only_that_message(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'failingProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    SimpleMessageChannelBuilder::create('failingProcessing', new FailOnceOnPayloadPollableChannel('cappuccino')),
+                    PollableChannelConfiguration::create('failingProcessing', RetryTemplateBuilder::fixedBackOff(1)->maxRetryAttempts(1)->build()),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso', 'latte'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('failingProcessing'))));
+        $this->assertSame(1, $this->amountOfRowsOn('outbox'));
+        $this->assertSame([true], $this->redeliveredFlagsOn('outbox'));
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['cappuccino'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('failingProcessing'))));
+        $this->assertSame(0, $this->amountOfRowsOn('outbox'));
+    }
+
+    public function test_single_message_is_forwarded_without_waiting_for_source_receive_timeout(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox')
+                        ->withReceiveTimeout(3000),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+
+        $startedAt = microtime(true);
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 10000));
+        $elapsedInMilliseconds = (microtime(true) - $startedAt) * 1000;
+
+        $this->assertCount(1, $this->receiveAllFrom($messaging->getMessageChannel('orderProcessing')));
+        $this->assertLessThan(3000, $elapsedInMilliseconds);
+    }
+
+    public function test_drained_message_honours_ignore_final_failure_strategy(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [
+                DbalConnectionFactory::class => $this->getConnectionFactory(),
+                $orderService,
+                'alwaysFailingDelivery' => new AlwaysFailOnPayloadChannelInterceptor('cappuccino'),
+            ],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing')
+                        ->withFinalFailureStrategy(FinalFailureStrategy::IGNORE),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                    SimpleChannelInterceptorBuilder::create('orderProcessing', 'alwaysFailingDelivery'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso', 'latte'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+        $this->assertNull($messaging->getMessageChannel('outbox')->receive());
+    }
+
+    public function test_drained_message_with_stop_failure_strategy_stops_consumer_without_message_loss(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [
+                DbalConnectionFactory::class => $this->getConnectionFactory(),
+                $orderService,
+                'alwaysFailingDelivery' => new AlwaysFailOnPayloadChannelInterceptor('cappuccino'),
+            ],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing')
+                        ->withFinalFailureStrategy(FinalFailureStrategy::STOP),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                    SimpleChannelInterceptorBuilder::create('orderProcessing', 'alwaysFailingDelivery'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $consumerStopped = false;
+        try {
+            $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+        } catch (RuntimeException) {
+            $consumerStopped = true;
+        }
+
+        $this->assertTrue($consumerStopped);
+        $this->assertCount(3, $this->receiveAllFrom($messaging->getMessageChannel('outbox')));
+    }
+
+    public function test_batch_forwarding_configuration_for_non_dbal_channel_fails_at_compile_time(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $this->expectException(ConfigurationException::class);
+
+        EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [$orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'inMemoryOutbox', 'inMemoryProcessing'),
+                    SimpleMessageChannelBuilder::createQueueChannel('inMemoryOutbox'),
+                    SimpleMessageChannelBuilder::createQueueChannel('inMemoryProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+    }
+
+    public function test_batch_forwarding_configuration_for_unknown_channel_fails_at_compile_time(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $this->expectException(ConfigurationException::class);
+
+        EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'misspelled_outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+    }
+
+    public function test_forwarded_message_does_not_carry_internal_collector_header(): void
+    {
+        foreach ([true, false] as $collectorEnabled) {
+            $orderService = new class () {
+                #[Asynchronous('orders')]
+                #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+                public function register(string $order): void
+                {
+                }
+            };
+
+            $messaging = EcotoneLite::bootstrapFlowTesting(
+                [$orderService::class],
+                [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+                ServiceConfiguration::createWithDefaults()
+                    ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                    ->withExtensionObjects([
+                        OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                        DbalBackedMessageChannelBuilder::create('outbox'),
+                        DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                        GlobalPollableChannelConfiguration::createWithDefaults()->withCollector($collectorEnabled),
+                    ]),
+                licenceKey: LicenceTesting::VALID_LICENCE,
+            );
+
+            $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+            $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+
+            $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+            $forwardedMessages = $this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'));
+            $this->assertCount(2, $forwardedMessages);
+            foreach ($forwardedMessages as $forwardedMessage) {
+                $this->assertFalse($forwardedMessage->getHeaders()->containsKey('collectorBypass'));
+            }
+        }
+    }
+
+    public function test_using_batched_outbox_directly_as_asynchronous_channel_fails_at_compile_time(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+
+            #[Asynchronous('outbox')]
+            #[CommandHandler('order.registerDirectly', endpointId: 'orderRegisterDirectlyEndpoint')]
+            public function registerDirectly(string $order): void
+            {
+            }
+        };
+
+        $this->expectException(ConfigurationException::class);
+
+        EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+    }
+
+    public function test_using_batched_outbox_directly_as_asynchronous_channel_without_combined_channel_fails_at_compile_time(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('outbox')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $this->expectException(ConfigurationException::class);
+
+        EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+    }
+
+    public function test_using_batched_outbox_as_output_channel_fails_at_compile_time(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+
+            #[CommandHandler('order.startFlow', outputChannelName: 'outbox')]
+            public function startFlow(string $order): string
+            {
+                return $order;
+            }
+        };
+
+        $this->expectException(ConfigurationException::class);
+
+        EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+    }
+
+    public function test_combined_channel_without_batch_forwarding_configuration_keeps_one_message_per_run(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    CombinedMessageChannel::create('orders', ['outbox', 'orderProcessing']),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertCount(1, $this->receiveAllFrom($messaging->getMessageChannel('orderProcessing')));
+        $this->assertCount(2, $this->receiveAllFrom($messaging->getMessageChannel('outbox')));
+    }
+
+    public function test_failed_forwarding_releases_all_messages_back_to_outbox(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'failingProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    SimpleMessageChannelBuilder::create('failingProcessing', new FailingPollableChannel()),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithFinishWhenNoMessages());
+
+        $this->assertCount(3, $this->receiveAllFrom($messaging->getMessageChannel('outbox')));
+    }
+
+    public function test_error_channel_is_not_involved_in_failed_forwarding_for_release_and_ignore_strategies(): void
+    {
+        foreach ([FinalFailureStrategy::RELEASE, FinalFailureStrategy::IGNORE] as $finalFailureStrategy) {
+            $orderService = new class () {
+                #[Asynchronous('orders')]
+                #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+                public function register(string $order): void
+                {
+                }
+            };
+
+            $messaging = EcotoneLite::bootstrapFlowTesting(
+                [$orderService::class],
+                [
+                    DbalConnectionFactory::class => $this->getConnectionFactory(),
+                    $orderService,
+                    'alwaysFailingDelivery' => new AlwaysFailOnPayloadChannelInterceptor('cappuccino'),
+                ],
+                ServiceConfiguration::createWithDefaults()
+                    ->withDefaultErrorChannel('customErrorChannel')
+                    ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                    ->withExtensionObjects([
+                        OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                        DbalBackedMessageChannelBuilder::create('outbox')
+                            ->withFinalFailureStrategy($finalFailureStrategy),
+                        DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                        SimpleMessageChannelBuilder::createQueueChannel('customErrorChannel'),
+                        SimpleChannelInterceptorBuilder::create('orderProcessing', 'alwaysFailingDelivery'),
+                    ]),
+                licenceKey: LicenceTesting::VALID_LICENCE,
+            );
+
+            $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+            $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+            $messaging->run('outbox', ExecutionPollingMetadata::createWithFinishWhenNoMessages());
+
+            $this->assertNull($messaging->getMessageChannel('customErrorChannel')->receive());
+            $this->assertSame(['espresso'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+            $expectedRemainingOnOutbox = $finalFailureStrategy === FinalFailureStrategy::RELEASE ? 1 : 0;
+            $this->assertCount($expectedRemainingOnOutbox, $this->receiveAllFrom($messaging->getMessageChannel('outbox')));
+        }
+    }
+
+    public function test_error_channel_is_not_involved_in_failed_forwarding_for_stop_strategy(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [
+                DbalConnectionFactory::class => $this->getConnectionFactory(),
+                $orderService,
+                'alwaysFailingDelivery' => new AlwaysFailOnPayloadChannelInterceptor('cappuccino'),
+            ],
+            ServiceConfiguration::createWithDefaults()
+                ->withDefaultErrorChannel('customErrorChannel')
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox')
+                        ->withFinalFailureStrategy(FinalFailureStrategy::STOP),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                    SimpleMessageChannelBuilder::createQueueChannel('customErrorChannel'),
+                    SimpleChannelInterceptorBuilder::create('orderProcessing', 'alwaysFailingDelivery'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $consumerStopped = false;
+        try {
+            $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+        } catch (RuntimeException) {
+            $consumerStopped = true;
+        }
+
+        $this->assertTrue($consumerStopped);
+        $this->assertNull($messaging->getMessageChannel('customErrorChannel')->receive());
+        $this->assertCount(2, $this->receiveAllFrom($messaging->getMessageChannel('outbox')));
+    }
+
+    public function test_messages_claimed_by_another_process_are_not_published_until_claim_expires(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [DbalConnectionFactory::class => $this->getConnectionFactory(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+
+        $this->claimSingleOutboxRowAsAnotherProcess('outbox', claimValidForSeconds: 3600);
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithFinishWhenNoMessages());
+
+        $this->assertCount(1, $this->receiveAllFrom($messaging->getMessageChannel('orderProcessing')));
+
+        $this->expireForeignOutboxClaims('outbox');
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithFinishWhenNoMessages());
+
+        $this->assertCount(1, $this->receiveAllFrom($messaging->getMessageChannel('orderProcessing')));
+        $this->assertNull($messaging->getMessageChannel('outbox')->receive());
+    }
+
+    private function claimSingleOutboxRowAsAnotherProcess(string $channelName, int $claimValidForSeconds): void
+    {
+        $connection = $this->getConnection();
+        $rowId = $connection->fetchOne('SELECT id FROM enqueue WHERE queue = ? AND delivery_id IS NULL ORDER BY published_at ASC LIMIT 1', [$channelName]);
+        $connection->executeStatement(
+            'UPDATE enqueue SET delivery_id = ?, redeliver_after = ? WHERE id = ?',
+            ['019890ab-0000-7000-8000-000000000001', time() + $claimValidForSeconds, $rowId],
+        );
+    }
+
+    private function expireForeignOutboxClaims(string $channelName): void
+    {
+        $this->getConnection()->executeStatement(
+            'UPDATE enqueue SET redeliver_after = ? WHERE queue = ? AND delivery_id IS NOT NULL',
+            [time() - 10, $channelName],
+        );
+    }
+
+    public function test_failed_delivery_of_single_message_releases_only_that_message_without_duplicates(): void
+    {
+        $messaging = $this->bootstrapWithFailingTargetInterceptor(RuntimeException::class);
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso', 'latte'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['cappuccino'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+        $this->assertNull($messaging->getMessageChannel('outbox')->receive());
+    }
+
+    public function test_connection_failure_during_delivery_is_recovered_without_duplicates(): void
+    {
+        $messaging = $this->bootstrapWithFailingTargetInterceptor(ConnectionException::class);
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso');
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte');
+        $messaging->sendCommandWithRoutingKey('order.register', 'cappuccino');
+
+        try {
+            $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+        } catch (ConnectionException) {
+        }
+
+        $this->assertSame(3, $this->amountOfRowsOn('outbox'));
+        $this->assertSame(0, $this->amountOfClaimedRowsOn('outbox'));
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso', 'latte', 'cappuccino'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+        $this->assertSame(0, $this->amountOfRowsOn('outbox'));
+    }
+
+    private function bootstrapWithFailingTargetInterceptor(string $exceptionClass): FlowTestSupport
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        return EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [
+                DbalConnectionFactory::class => $this->getConnectionFactory(),
+                $orderService,
+                'failingDeliveryInterceptor' => new FailOnceOnPayloadChannelInterceptor('cappuccino', $exceptionClass),
+            ],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    DbalBackedMessageChannelBuilder::create('orderProcessing'),
+                    SimpleChannelInterceptorBuilder::create('orderProcessing', 'failingDeliveryInterceptor'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+    }
+
+    private function amountOfRowsOn(string $channelName): int
+    {
+        return (int) $this->getConnection()->fetchOne('SELECT COUNT(*) FROM enqueue WHERE queue = ?', [$channelName]);
+    }
+
+    private function amountOfClaimedRowsOn(string $channelName): int
+    {
+        return (int) $this->getConnection()->fetchOne('SELECT COUNT(*) FROM enqueue WHERE queue = ? AND delivery_id IS NOT NULL', [$channelName]);
+    }
+
+    private function amountOfPendingRowsOn(string $channelName): int
+    {
+        return (int) $this->getConnection()->fetchOne('SELECT COUNT(*) FROM enqueue WHERE queue = ? AND delivery_id IS NULL', [$channelName]);
+    }
+
+    /**
+     * @return bool[]
+     */
+    private function redeliveredFlagsOn(string $channelName): array
+    {
+        return array_map(
+            fn ($redelivered) => (bool) $redelivered,
+            $this->getConnection()->fetchFirstColumn('SELECT redelivered FROM enqueue WHERE queue = ? ORDER BY published_at ASC', [$channelName]),
+        );
+    }
+
+    /**
+     * @param \Ecotone\Messaging\Message[] $messages
+     * @return string[]
+     */
+    private function payloadsOf(array $messages): array
+    {
+        return array_map(fn ($message) => $message->getPayload(), $messages);
+    }
+
+    private function receiveAllFrom(PollableChannel $channel): array
+    {
+        $messages = [];
+        while ($message = $channel->receive()) {
+            $messages[] = $message;
+        }
+
+        return $messages;
+    }
+}
diff --git a/packages/Dbal/tests/Integration/MultiTenant/BatchForwardingMultiTenantTest.php b/packages/Dbal/tests/Integration/MultiTenant/BatchForwardingMultiTenantTest.php
new file mode 100644
index 000000000..196566dd4
--- /dev/null
+++ b/packages/Dbal/tests/Integration/MultiTenant/BatchForwardingMultiTenantTest.php
@@ -0,0 +1,105 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Integration\MultiTenant;
+
+use Ecotone\Dbal\DbalBackedMessageChannelBuilder;
+use Ecotone\Dbal\MultiTenant\MultiTenantConfiguration;
+use Ecotone\Dbal\OutboxForwardingMessageChannel;
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Messaging\Attribute\Asynchronous;
+use Ecotone\Messaging\Channel\SimpleMessageChannelBuilder;
+use Ecotone\Messaging\Config\ModulePackageList;
+use Ecotone\Messaging\Config\ServiceConfiguration;
+use Ecotone\Messaging\Endpoint\ExecutionPollingMetadata;
+use Ecotone\Modelling\Attribute\CommandHandler;
+use Ecotone\Test\LicenceTesting;
+use Interop\Queue\ConnectionFactory;
+use Test\Ecotone\Dbal\DbalMessagingTestCase;
+
+/**
+ * licence Apache-2.0
+ * @internal
+ */
+final class BatchForwardingMultiTenantTest extends DbalMessagingTestCase
+{
+    public function test_each_run_publishes_batch_from_next_tenant_outbox_in_round_robin(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $messaging = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [
+                $orderService,
+                'tenant_a_connection' => $this->connectionForTenantA(),
+                'tenant_b_connection' => $this->connectionForTenantB(),
+            ],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    MultiTenantConfiguration::create(
+                        'tenant',
+                        ['tenant_a' => 'tenant_a_connection', 'tenant_b' => 'tenant_b_connection'],
+                    ),
+                    OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing'),
+                    DbalBackedMessageChannelBuilder::create('outbox'),
+                    SimpleMessageChannelBuilder::createQueueChannel('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $messaging->sendCommandWithRoutingKey('order.register', 'espresso', metadata: ['tenant' => 'tenant_a']);
+        $messaging->sendCommandWithRoutingKey('order.register', 'latte', metadata: ['tenant' => 'tenant_a']);
+        $messaging->sendCommandWithRoutingKey('order.register', 'flat white', metadata: ['tenant' => 'tenant_b']);
+
+        $this->assertSame(2, $this->amountOfOutboxRowsFor($this->connectionForTenantA()));
+        $this->assertSame(1, $this->amountOfOutboxRowsFor($this->connectionForTenantB()));
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 1, maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['espresso', 'latte'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+        $this->assertSame(0, $this->amountOfOutboxRowsFor($this->connectionForTenantA()));
+        $this->assertSame(1, $this->amountOfOutboxRowsFor($this->connectionForTenantB()));
+
+        $messaging->run('outbox', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 1, maxExecutionTimeInMilliseconds: 5000));
+
+        $this->assertSame(['flat white'], $this->payloadsOf($this->receiveAllFrom($messaging->getMessageChannel('orderProcessing'))));
+        $this->assertSame(0, $this->amountOfOutboxRowsFor($this->connectionForTenantB()));
+    }
+
+    private function amountOfOutboxRowsFor(ConnectionFactory $connectionFactory): int
+    {
+        $connection = $connectionFactory->createContext()->getDbalConnection();
+        if (! self::checkIfTableExists($connection, 'enqueue')) {
+            return 0;
+        }
+
+        return (int) $connection->fetchOne('SELECT COUNT(*) FROM enqueue WHERE queue = ?', ['outbox']);
+    }
+
+    /**
+     * @param \Ecotone\Messaging\Message[] $messages
+     * @return string[]
+     */
+    private function payloadsOf(array $messages): array
+    {
+        return array_map(fn ($message) => $message->getPayload(), $messages);
+    }
+
+    private function receiveAllFrom(\Ecotone\Messaging\PollableChannel $channel): array
+    {
+        $messages = [];
+        while ($message = $channel->receive()) {
+            $messages[] = $message;
+        }
+
+        return $messages;
+    }
+}
diff --git a/packages/Ecotone/src/Messaging/Channel/CombinedMessageChannel.php b/packages/Ecotone/src/Messaging/Channel/CombinedMessageChannel.php
index bdf05f368..73498d957 100644
--- a/packages/Ecotone/src/Messaging/Channel/CombinedMessageChannel.php
+++ b/packages/Ecotone/src/Messaging/Channel/CombinedMessageChannel.php
@@ -9,9 +9,9 @@
 /**
  * licence Apache-2.0
  */
-final class CombinedMessageChannel
+class CombinedMessageChannel
 {
-    private function __construct(private string $referenceName, private array $combinedChannels)
+    protected function __construct(private string $referenceName, private array $combinedChannels)
     {
         Assert::notNull($referenceName, 'Reference name can not be null');
         Assert::notNullAndEmpty($this->combinedChannels, 'Combined channels can not be empty');
diff --git a/packages/Ecotone/src/Messaging/Channel/OutboxForwardingChannel.php b/packages/Ecotone/src/Messaging/Channel/OutboxForwardingChannel.php
new file mode 100644
index 000000000..a2d36ff64
--- /dev/null
+++ b/packages/Ecotone/src/Messaging/Channel/OutboxForwardingChannel.php
@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Messaging\Channel;
+
+/**
+ * Combined Message Channel whose source acts as an outbox published by a dedicated forwarding endpoint.
+ * Implementations own the storage specific configuration, while Messaging core guards that the source
+ * channel is claimed by a forwarding module and not consumed or reused as a plain combined channel.
+ *
+ * licence Enterprise
+ */
+interface OutboxForwardingChannel
+{
+    public function getReferenceName(): string;
+
+    public function getSourceChannelName(): string;
+
+    public function getEmbeddedSourceChannelBuilder(): ?MessageChannelBuilder;
+}
diff --git a/packages/Ecotone/src/Messaging/Channel/PollableChannel/Serialization/OutboundMessageConverter.php b/packages/Ecotone/src/Messaging/Channel/PollableChannel/Serialization/OutboundMessageConverter.php
index 13924da72..2d00572cf 100644
--- a/packages/Ecotone/src/Messaging/Channel/PollableChannel/Serialization/OutboundMessageConverter.php
+++ b/packages/Ecotone/src/Messaging/Channel/PollableChannel/Serialization/OutboundMessageConverter.php
@@ -13,6 +13,7 @@
 use Ecotone\Messaging\Scheduling\DatePoint;
 use Ecotone\Messaging\Scheduling\Duration;
 use Ecotone\Messaging\Scheduling\TimeSpan;
+use Ecotone\Messaging\Support\MessageBuilder;
 
 /**
  * licence Apache-2.0
@@ -29,6 +30,51 @@ public function __construct(
     ) {
     }
 
+    /**
+     * Behaves identically to preparing a full Message built from the entry: string payloads carrying own
+     * message id and timestamp skip the Message construction, as no payload conversion applies to them.
+     *
+     * @param array{payload: mixed, headers: array<string, mixed>} $batchEntry
+     */
+    public function prepareFromBatchEntry(array $batchEntry, ConversionService $conversionService): OutboundMessage
+    {
+        $payload = $batchEntry['payload'];
+        $headers = $batchEntry['headers'];
+
+        if (! is_string($payload) || ! isset($headers[MessageHeaders::MESSAGE_ID], $headers[MessageHeaders::TIMESTAMP])) {
+            return $this->prepare(
+                MessageBuilder::withPayload($payload)->setMultipleHeaders($headers)->build(),
+                $conversionService,
+            );
+        }
+
+        $applicationHeaders = MessageHeaders::unsetAggregateKeys($headers);
+        $applicationHeaders = MessageHeaders::unsetEnqueueMetadata($applicationHeaders);
+        $applicationHeaders = $this->headerMapper->mapFromMessageHeaders($applicationHeaders, $conversionService);
+        $applicationHeaders[MessageHeaders::MESSAGE_ID] = $headers[MessageHeaders::MESSAGE_ID];
+        $applicationHeaders[MessageHeaders::TIMESTAMP] = $headers[MessageHeaders::TIMESTAMP];
+
+        if (isset($headers[MessageHeaders::ROUTING_SLIP])) {
+            $applicationHeaders[MessageHeaders::ROUTING_SLIP] = $headers[MessageHeaders::ROUTING_SLIP];
+        }
+        $contentType = isset($headers[MessageHeaders::CONTENT_TYPE]) ? MediaType::parseMediaType($headers[MessageHeaders::CONTENT_TYPE])->toString() : null;
+        $applicationHeaders[MessageHeaders::CONTENT_TYPE] = $contentType;
+
+        $deliveryDelay = $this->normalizeDeliveryDelay(
+            array_key_exists(MessageHeaders::DELIVERY_DELAY, $headers) ? $headers[MessageHeaders::DELIVERY_DELAY] : $this->defaultDeliveryDelay,
+            $headers[MessageHeaders::TIMESTAMP],
+        );
+
+        return new OutboundMessage(
+            $payload,
+            array_merge($applicationHeaders, $this->staticHeadersToAdd),
+            $contentType,
+            $deliveryDelay,
+            array_key_exists(MessageHeaders::TIME_TO_LIVE, $headers) ? $headers[MessageHeaders::TIME_TO_LIVE] : $this->defaultTimeToLive,
+            array_key_exists(MessageHeaders::PRIORITY, $headers) ? $headers[MessageHeaders::PRIORITY] : $this->defaultPriority,
+        );
+    }
+
     public function prepare(Message $messageToConvert, ConversionService $conversionService): OutboundMessage
     {
         $messagePayload = $messageToConvert->getPayload();
@@ -97,10 +143,25 @@ public function prepare(Message $messageToConvert, ConversionService $conversion
         }
         $applicationHeaders[MessageHeaders::CONTENT_TYPE] = $sourceMediaType?->toString();
 
-        $deliveryDelay = $messageToConvert->getHeaders()->containsKey(MessageHeaders::DELIVERY_DELAY) ? $messageToConvert->getHeaders()->get(MessageHeaders::DELIVERY_DELAY) : $this->defaultDeliveryDelay;
+        $deliveryDelay = $this->normalizeDeliveryDelay(
+            $messageToConvert->getHeaders()->containsKey(MessageHeaders::DELIVERY_DELAY) ? $messageToConvert->getHeaders()->get(MessageHeaders::DELIVERY_DELAY) : $this->defaultDeliveryDelay,
+            $messageToConvert->getHeaders()->getTimestamp(),
+        );
 
+        return new OutboundMessage(
+            $messagePayload,
+            array_merge($applicationHeaders, $this->staticHeadersToAdd),
+            $applicationHeaders[MessageHeaders::CONTENT_TYPE],
+            $deliveryDelay,
+            $messageToConvert->getHeaders()->containsKey(MessageHeaders::TIME_TO_LIVE) ? $messageToConvert->getHeaders()->get(MessageHeaders::TIME_TO_LIVE) : $this->defaultTimeToLive,
+            $messageToConvert->getHeaders()->containsKey(MessageHeaders::PRIORITY) ? $messageToConvert->getHeaders()->get(MessageHeaders::PRIORITY) : $this->defaultPriority,
+        );
+    }
+
+    private function normalizeDeliveryDelay(mixed $deliveryDelay, int $messageTimestamp): ?int
+    {
         if ($deliveryDelay instanceof DateTimeInterface) {
-            $deliveryDelay = DatePoint::createFromInterface($deliveryDelay)->durationSince(DatePoint::createFromTimestamp($messageToConvert->getHeaders()->getTimestamp()));
+            $deliveryDelay = DatePoint::createFromInterface($deliveryDelay)->durationSince(DatePoint::createFromTimestamp($messageTimestamp));
         }
 
         if ($deliveryDelay instanceof Duration) {
@@ -115,14 +176,7 @@ public function prepare(Message $messageToConvert, ConversionService $conversion
             $deliveryDelay = null;
         }
 
-        return new OutboundMessage(
-            $messagePayload,
-            array_merge($applicationHeaders, $this->staticHeadersToAdd),
-            $applicationHeaders[MessageHeaders::CONTENT_TYPE],
-            $deliveryDelay,
-            $messageToConvert->getHeaders()->containsKey(MessageHeaders::TIME_TO_LIVE) ? $messageToConvert->getHeaders()->get(MessageHeaders::TIME_TO_LIVE) : $this->defaultTimeToLive,
-            $messageToConvert->getHeaders()->containsKey(MessageHeaders::PRIORITY) ? $messageToConvert->getHeaders()->get(MessageHeaders::PRIORITY) : $this->defaultPriority,
-        );
+        return $deliveryDelay;
     }
 
     private function doesRequireConversion(
diff --git a/packages/Ecotone/src/Messaging/Config/Configuration.php b/packages/Ecotone/src/Messaging/Config/Configuration.php
index 085be7f9b..560d24fb6 100644
--- a/packages/Ecotone/src/Messaging/Config/Configuration.php
+++ b/packages/Ecotone/src/Messaging/Config/Configuration.php
@@ -67,6 +67,12 @@ public function registerChannelInterceptor(ChannelInterceptorBuilder $channelInt
      */
     public function registerAsynchronousEndpoint(array|string $asynchronousChannelNames, string $targetEndpointId): Configuration;
 
+    /**
+     * Marks given Message Channel as batched forwarding outbox, which removes its standard Message Channel consumer
+     * and ensures the Channel is not used as execution or output channel.
+     */
+    public function registerBatchForwardingSourceChannel(string $channelName): Configuration;
+
     /**
      * @param MethodInterceptorBuilder $methodInterceptor
      * @return Configuration
diff --git a/packages/Ecotone/src/Messaging/Config/MessagingSystemConfiguration.php b/packages/Ecotone/src/Messaging/Config/MessagingSystemConfiguration.php
index 58b4652e6..06c0f2efc 100644
--- a/packages/Ecotone/src/Messaging/Config/MessagingSystemConfiguration.php
+++ b/packages/Ecotone/src/Messaging/Config/MessagingSystemConfiguration.php
@@ -12,8 +12,10 @@
 use Ecotone\Messaging\Attribute\Asynchronous;
 use Ecotone\Messaging\Attribute\AsynchronousRunningEndpoint;
 use Ecotone\Messaging\Channel\ChannelInterceptorBuilder;
+use Ecotone\Messaging\Channel\CombinedMessageChannel;
 use Ecotone\Messaging\Channel\EventDrivenChannelInterceptorAdapter;
 use Ecotone\Messaging\Channel\MessageChannelBuilder;
+use Ecotone\Messaging\Channel\OutboxForwardingChannel;
 use Ecotone\Messaging\Channel\PollableChannelInterceptorAdapter;
 use Ecotone\Messaging\Channel\SimpleMessageChannelBuilder;
 use Ecotone\Messaging\Config\Annotation\AnnotationModuleRetrievingService;
@@ -141,6 +143,18 @@ final class MessagingSystemConfiguration implements Configuration
     private array $messageConverterReferenceNames = [];
     private ?ModuleReferenceSearchService $moduleReferenceSearchService;
     private array $asynchronousEndpoints = [];
+    /**
+     * @var array<string, bool>
+     */
+    private array $batchForwardingSourceChannels = [];
+    /**
+     * @var string[]
+     */
+    private array $declaredBatchForwardingChannels = [];
+    /**
+     * @var array<string, string[]>
+     */
+    private array $plainCombinedChannelUsages = [];
     private ServiceConfiguration $applicationConfiguration;
     /**
      * @var string[]
@@ -219,6 +233,17 @@ function ($extensionObject) {
 
         $this->isRunningForTest = ExtensionObjectResolver::contains(TestConfiguration::class, $extensionObjects);
 
+        foreach (ExtensionObjectResolver::resolve(CombinedMessageChannel::class, $extensionObjects) as $combinedMessageChannel) {
+            if ($combinedMessageChannel instanceof OutboxForwardingChannel) {
+                $this->declaredBatchForwardingChannels[] = $combinedMessageChannel->getSourceChannelName();
+                if ($combinedMessageChannel->getEmbeddedSourceChannelBuilder() !== null) {
+                    $extensionObjects[] = $combinedMessageChannel->getEmbeddedSourceChannelBuilder();
+                }
+            } else {
+                $this->plainCombinedChannelUsages[$combinedMessageChannel->getReferenceName()] = $combinedMessageChannel->getCombinedChannels();
+            }
+        }
+
         $extensionObjects[] = $serviceConfiguration;
 
         if ($serviceConfiguration->getLicenceKey() !== null) {
@@ -409,6 +434,21 @@ private function configureAsynchronousEndpoints(InterfaceToCallRegistry $interfa
         /** @var array<string, AttributeDefinition[]> $asyncHandlerAnnotations */
         $asyncHandlerAnnotations = [];
 
+        if ($this->batchForwardingSourceChannels !== [] && ! $this->isRunningForEnterpriseLicence) {
+            throw LicensingException::create(sprintf('Batch forwarding for Message Channel `%s` is available only with Ecotone Enterprise licence.', array_key_first($this->batchForwardingSourceChannels)));
+        }
+        foreach ($this->declaredBatchForwardingChannels as $declaredBatchForwardingChannel) {
+            if (! isset($this->batchForwardingSourceChannels[$declaredBatchForwardingChannel])) {
+                throw ConfigurationException::create("Outbox forwarding was configured with source Message Channel `{$declaredBatchForwardingChannel}`, yet no module enabled it for that channel. Outbox forwarding requires a Dbal backed Message Channel as the source and the Dbal package enabled.");
+            }
+            foreach ($this->plainCombinedChannelUsages as $combinedChannelReferenceName => $referencedChannels) {
+                if (in_array($declaredBatchForwardingChannel, $referencedChannels, true)) {
+                    throw ConfigurationException::create("Message Channel `{$declaredBatchForwardingChannel}` is the outbox source of an Outbox forwarding Message Channel and can not be reused inside Combined Message Channel `{$combinedChannelReferenceName}`. Define that flow with OutboxForwardingMessageChannel as well.");
+                }
+            }
+        }
+        $this->verifyBatchForwardingSourceChannelsAreNotUsedForExecution($this->batchForwardingSourceChannels);
+
         foreach ($this->asynchronousEndpoints as $targetEndpointId => $asynchronousMessageChannels) {
             $asynchronousMessageChannel = array_shift($asynchronousMessageChannels);
             if (! isset($this->channelBuilders[$asynchronousMessageChannel]) && ! isset($this->defaultChannelBuilders[$asynchronousMessageChannel])) {
@@ -536,6 +576,14 @@ private function configureAsynchronousEndpoints(InterfaceToCallRegistry $interfa
              * This is Bridge that will fetch the message and make use of routing_slip to target it
              * message handler.
              */
+            if (isset($this->batchForwardingSourceChannels[$asynchronousChannel])) {
+                if (! isset($this->channelAdapters[$asynchronousChannel])) {
+                    unset($this->pollingMetadata[$asynchronousChannel]);
+                }
+
+                continue;
+            }
+
             $this->messageHandlerBuilders[$asynchronousChannel] = BridgeBuilder::create()
                 ->withInputChannelName($asynchronousChannel)
                 ->withEndpointId($asynchronousChannel);
@@ -544,6 +592,29 @@ private function configureAsynchronousEndpoints(InterfaceToCallRegistry $interfa
         $this->asynchronousEndpoints = [];
     }
 
+    /**
+     * @param array<string, bool> $batchForwardingSourceChannels
+     */
+    private function verifyBatchForwardingSourceChannelsAreNotUsedForExecution(array $batchForwardingSourceChannels): void
+    {
+        if ($batchForwardingSourceChannels === []) {
+            return;
+        }
+
+        foreach ($this->asynchronousEndpoints as $targetEndpointId => $asynchronousMessageChannels) {
+            $executionChannel = $asynchronousMessageChannels[array_key_last($asynchronousMessageChannels)];
+            if (isset($batchForwardingSourceChannels[$executionChannel])) {
+                throw ConfigurationException::create("Channel `{$executionChannel}` is a batched forwarding outbox, which only pushes messages forward to the next channel. It can not be used as execution channel for endpoint `{$targetEndpointId}`. Point the endpoint at the Combined Message Channel instead.");
+            }
+        }
+
+        foreach ($this->messageHandlerBuilders as $messageHandlerBuilder) {
+            if ($messageHandlerBuilder instanceof MessageHandlerBuilderWithOutputChannel && isset($batchForwardingSourceChannels[$messageHandlerBuilder->getOutputMessageChannelName()])) {
+                throw ConfigurationException::create("Channel `{$messageHandlerBuilder->getOutputMessageChannelName()}` is a batched forwarding outbox, which only pushes messages forward to the next channel. It can not be used as output channel of {$messageHandlerBuilder}. Point the output at the Combined Message Channel instead.");
+            }
+        }
+    }
+
     /**
      * @return void
      */
@@ -756,6 +827,14 @@ public function registerAroundMethodInterceptor(AroundInterceptorBuilder $around
     /**
      * @inheritDoc
      */
+    public function registerBatchForwardingSourceChannel(string $channelName): Configuration
+    {
+        Assert::isTrue(! isset($this->batchForwardingSourceChannels[$channelName]), "Batch forwarding for Message Channel `{$channelName}` is already registered.");
+        $this->batchForwardingSourceChannels[$channelName] = true;
+
+        return $this;
+    }
+
     public function registerAsynchronousEndpoint(array|string $asynchronousChannelNames, string $targetEndpointId): Configuration
     {
         $this->asynchronousEndpoints[$targetEndpointId] = is_string($asynchronousChannelNames) ? [$asynchronousChannelNames] : $asynchronousChannelNames;
diff --git a/packages/Ecotone/src/Messaging/Config/ModuleClassList.php b/packages/Ecotone/src/Messaging/Config/ModuleClassList.php
index 1046e033c..227e45b5e 100644
--- a/packages/Ecotone/src/Messaging/Config/ModuleClassList.php
+++ b/packages/Ecotone/src/Messaging/Config/ModuleClassList.php
@@ -9,6 +9,7 @@
 use Ecotone\Amqp\Publisher\AmqpMessagePublisherModule;
 use Ecotone\Amqp\Transaction\AmqpTransactionModule;
 use Ecotone\DataProtection\Configuration\DataProtectionModule;
+use Ecotone\Dbal\BatchForwarding\DbalBatchForwardingModule;
 use Ecotone\Dbal\Configuration\DbalConnectionModule;
 use Ecotone\Dbal\Configuration\DbalPublisherModule;
 use Ecotone\Dbal\Database\DatabaseSetupModule;
@@ -155,6 +156,7 @@ class ModuleClassList
     ];
 
     public const DBAL_MODULES = [
+        DbalBatchForwardingModule::class,
         DbalConnectionModule::class,
         DbalDeadLetterModule::class,
         ObjectManagerModule::class,
diff --git a/packages/Ecotone/src/Messaging/Endpoint/InboundChannelAdapter/InboundChannelAdapterBuilder.php b/packages/Ecotone/src/Messaging/Endpoint/InboundChannelAdapter/InboundChannelAdapterBuilder.php
index 442ab8b6c..8ceea41b3 100644
--- a/packages/Ecotone/src/Messaging/Endpoint/InboundChannelAdapter/InboundChannelAdapterBuilder.php
+++ b/packages/Ecotone/src/Messaging/Endpoint/InboundChannelAdapter/InboundChannelAdapterBuilder.php
@@ -30,6 +30,7 @@ class InboundChannelAdapterBuilder extends InterceptedChannelAdapterBuilder
     private string $referenceName;
     private string $requestChannelName;
     private ?object $directObject = null;
+    private bool $continuousPolling = false;
 
     private function __construct(string $requestChannelName, string $referenceName, private InterfaceToCall $interfaceToCall)
     {
@@ -115,9 +116,16 @@ public function withRequiredInterceptorNames(iterable $interceptorNames): self
         return $this;
     }
 
+    public function withContinuousPolling(bool $continuousPolling = true): self
+    {
+        $this->continuousPolling = $continuousPolling;
+
+        return $this;
+    }
+
     protected function withContinuesPolling(): bool
     {
-        return false;
+        return $this->continuousPolling;
     }
 
     public function compile(MessagingContainerBuilder $builder): Definition
diff --git a/packages/Ecotone/src/Messaging/Endpoint/InterceptedChannelAdapterBuilder.php b/packages/Ecotone/src/Messaging/Endpoint/InterceptedChannelAdapterBuilder.php
index e8c0f9811..7e6039cb9 100644
--- a/packages/Ecotone/src/Messaging/Endpoint/InterceptedChannelAdapterBuilder.php
+++ b/packages/Ecotone/src/Messaging/Endpoint/InterceptedChannelAdapterBuilder.php
@@ -44,7 +44,7 @@ abstract class InterceptedChannelAdapterBuilder implements ChannelAdapterConsume
 
     protected function withContinuesPolling(): bool
     {
-        return true;
+        return false;
     }
 
     abstract protected function getInterceptedInterface(InterfaceToCallRegistry $interfaceToCallRegistry): InterfaceToCall;
@@ -76,7 +76,7 @@ public function registerConsumer(MessagingContainerBuilder $builder): void
             new Reference(AsyncHandlerAnnotationRegistry::class),
             new Reference(AsyncEndpointAnnotationContext::class),
         ]);
-        $builder->registerPollingEndpoint($this->endpointId, $consumerRunner);
+        $builder->registerPollingEndpoint($this->endpointId, $consumerRunner, $this->withContinuesPolling());
     }
 
     private function getErrorInterceptorReference(MessagingContainerBuilder $builder): AroundInterceptorBuilder
diff --git a/packages/Ecotone/src/Messaging/MessageConverter/DefaultHeaderMapper.php b/packages/Ecotone/src/Messaging/MessageConverter/DefaultHeaderMapper.php
index 89bc167ec..546208737 100644
--- a/packages/Ecotone/src/Messaging/MessageConverter/DefaultHeaderMapper.php
+++ b/packages/Ecotone/src/Messaging/MessageConverter/DefaultHeaderMapper.php
@@ -120,6 +120,14 @@ private function mapHeaders(array $mappingHeaders, array $sourceHeaders, Convers
                 continue;
             }
 
+            if ($mappedHeader === '.*') {
+                foreach ($convertedSourceHeaders as $sourceHeaderName => $value) {
+                    $targetHeaders = $this->convertToStoreableFormat($sourceHeaderName, $value, $targetHeaders, $conversionService);
+                }
+
+                continue;
+            }
+
             foreach ($convertedSourceHeaders as $sourceHeaderName => $value) {
                 if (preg_match("#{$mappedHeader}#", $sourceHeaderName)) {
                     $targetHeaders = $this->convertToStoreableFormat($sourceHeaderName, $value, $targetHeaders, $conversionService);
@@ -138,7 +146,7 @@ private function mapHeaders(array $mappingHeaders, array $sourceHeaders, Convers
      */
     private function isScalarType($headerValue): bool
     {
-        return (Type::createFromVariable($headerValue))->isScalar();
+        return is_scalar($headerValue) || (Type::createFromVariable($headerValue))->isScalar();
     }
 
     /**
diff --git a/packages/Enqueue/src/EnqueueInboundChannelAdapterBuilder.php b/packages/Enqueue/src/EnqueueInboundChannelAdapterBuilder.php
index 75871b5f1..a999854ff 100644
--- a/packages/Enqueue/src/EnqueueInboundChannelAdapterBuilder.php
+++ b/packages/Enqueue/src/EnqueueInboundChannelAdapterBuilder.php
@@ -168,6 +168,11 @@ public function withDeclareOnStartup(bool $declareOnStartup): self
         return $this;
     }
 
+    public function getFinalFailureStrategy(): FinalFailureStrategy
+    {
+        return $this->finalFailureStrategy;
+    }
+
     /**
      * @return string
      */
diff --git a/packages/Enqueue/src/EnqueueOutboundChannelAdapter.php b/packages/Enqueue/src/EnqueueOutboundChannelAdapter.php
index 289eae2d2..addbe7121 100644
--- a/packages/Enqueue/src/EnqueueOutboundChannelAdapter.php
+++ b/packages/Enqueue/src/EnqueueOutboundChannelAdapter.php
@@ -101,6 +101,14 @@ protected function prepareOutboundMessage(Message $message): OutboundMessage
         return $this->outboundMessageConverter->prepare($message, $this->conversionService);
     }
 
+    /**
+     * @param array{payload: mixed, headers: array<string, mixed>} $entry
+     */
+    protected function prepareOutboundMessageFromBatchEntry(array $entry): OutboundMessage
+    {
+        return $this->outboundMessageConverter->prepareFromBatchEntry($entry, $this->conversionService);
+    }
+
     protected function sendSingleMessage(Message $message, Context $context): void
     {
         $outboundMessage                       = $this->prepareOutboundMessage($message);
diff --git a/packages/JmsConverter/composer.json b/packages/JmsConverter/composer.json
index 004ad531f..794f65f32 100644
--- a/packages/JmsConverter/composer.json
+++ b/packages/JmsConverter/composer.json
@@ -54,7 +54,7 @@
         "symfony/cache": "^6.4|^7.0|^8.0"
     },
     "require-dev": {
-        "ecotone/pdo-event-sourcing": "~1.320.0",
+        "ecotone/pdo-event-sourcing": "~1.322.2",
         "phpunit/phpunit": "^10.5|^11.0",
         "phpstan/phpstan": "^1.8"
     },
diff --git a/packages/Kafka/src/Channel/KafkaMessageChannelBuilder.php b/packages/Kafka/src/Channel/KafkaMessageChannelBuilder.php
index 90de43b80..fcabf14ba 100644
--- a/packages/Kafka/src/Channel/KafkaMessageChannelBuilder.php
+++ b/packages/Kafka/src/Channel/KafkaMessageChannelBuilder.php
@@ -119,7 +119,7 @@ public function withDefaultConversionMediaType(string $mediaType): self
         return $this;
     }
 
-    public function withAsyncPublishing(bool $enabled = true, ?int $timeoutInMilliseconds = null): self
+    public function withHighThroughputPublishing(bool $enabled = true, ?int $timeoutInMilliseconds = null): self
     {
         Assert::isTrue($timeoutInMilliseconds === null || $timeoutInMilliseconds > 0, 'Async publishing timeout must be a positive amount of milliseconds.');
         $this->asyncPublishing = $enabled;
@@ -130,7 +130,7 @@ public function withAsyncPublishing(bool $enabled = true, ?int $timeoutInMillise
         return $this;
     }
 
-    public function isAsyncPublishingEnabled(): bool
+    public function isHighThroughputPublishingEnabled(): bool
     {
         return $this->asyncPublishing;
     }
diff --git a/packages/Kafka/src/Configuration/KafkaModule.php b/packages/Kafka/src/Configuration/KafkaModule.php
index f4bd3a87d..ad1883bfe 100644
--- a/packages/Kafka/src/Configuration/KafkaModule.php
+++ b/packages/Kafka/src/Configuration/KafkaModule.php
@@ -113,7 +113,7 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO
                     MessagePublisher::class . '::' . $extensionObject->getMessageChannelName(),
                 )
                     ->withHeaderMapper($extensionObject->getHeaderMapper())
-                    ->withAsyncPublishing($extensionObject->isAsyncPublishingEnabled(), $extensionObject->getAsyncPublishingTimeout());
+                    ->withAsyncPublishing($extensionObject->isHighThroughputPublishingEnabled(), $extensionObject->getAsyncPublishingTimeout());
             }
         }
 
diff --git a/packages/Kafka/tests/Integration/AsyncPublishingTest.php b/packages/Kafka/tests/Integration/AsyncPublishingTest.php
index 54fc2a094..40a65ea87 100644
--- a/packages/Kafka/tests/Integration/AsyncPublishingTest.php
+++ b/packages/Kafka/tests/Integration/AsyncPublishingTest.php
@@ -99,7 +99,7 @@ public function test_async_publishing_via_message_channel_requires_enterprise_li
                         'async_orders',
                         topicName: $uniqueId = Uuid::v7()->toRfc4122(),
                         messageGroupId: $uniqueId,
-                    )->withAsyncPublishing(),
+                    )->withHighThroughputPublishing(),
                 ]),
         );
     }
@@ -220,10 +220,10 @@ private function bootstrapEcotone(string $channelName, object $orderService, Kaf
             $channelName,
             topicName: $uniqueId = Uuid::v7()->toRfc4122(),
             messageGroupId: $uniqueId,
-        )->withAsyncPublishing();
+        )->withHighThroughputPublishing();
 
         if ($asyncPublishingTimeout !== null) {
-            $channelBuilder = $channelBuilder->withAsyncPublishing(timeoutInMilliseconds: $asyncPublishingTimeout);
+            $channelBuilder = $channelBuilder->withHighThroughputPublishing(timeoutInMilliseconds: $asyncPublishingTimeout);
         }
 
         return EcotoneLite::bootstrapFlowTesting(
diff --git a/packages/Kafka/tests/Integration/CombinedChannelForwardingTest.php b/packages/Kafka/tests/Integration/CombinedChannelForwardingTest.php
new file mode 100644
index 000000000..ba0b5dc98
--- /dev/null
+++ b/packages/Kafka/tests/Integration/CombinedChannelForwardingTest.php
@@ -0,0 +1,56 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Kafka\Integration;
+
+use Ecotone\Dbal\OutboxForwardingMessageChannel;
+use Ecotone\Kafka\Channel\KafkaMessageChannelBuilder;
+use Ecotone\Kafka\Configuration\KafkaBrokerConfiguration;
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Messaging\Attribute\Asynchronous;
+use Ecotone\Messaging\Channel\SimpleMessageChannelBuilder;
+use Ecotone\Messaging\Config\ConfigurationException;
+use Ecotone\Messaging\Config\ModulePackageList;
+use Ecotone\Messaging\Config\ServiceConfiguration;
+use Ecotone\Modelling\Attribute\CommandHandler;
+use Ecotone\Test\LicenceTesting;
+use PHPUnit\Framework\Attributes\RunTestsInSeparateProcesses;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Uid\Uuid;
+use Test\Ecotone\Kafka\ConnectionTestCase;
+
+/**
+ * licence Enterprise
+ * @internal
+ */
+#[RunTestsInSeparateProcesses]
+final class CombinedChannelForwardingTest extends TestCase
+{
+    public function test_batch_forwarding_configuration_for_kafka_source_channel_fails_at_compile_time(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.register', endpointId: 'orderRegisterEndpoint')]
+            public function register(string $order): void
+            {
+            }
+        };
+
+        $this->expectException(ConfigurationException::class);
+
+        $uniqueId = Uuid::v7()->toRfc4122();
+        EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [KafkaBrokerConfiguration::class => ConnectionTestCase::getConnection(), $orderService],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::KAFKA_PACKAGE]))
+                ->withExtensionObjects([
+                    OutboxForwardingMessageChannel::create('orders', 'kafkaOutbox', 'orderProcessing'),
+                    KafkaMessageChannelBuilder::create('kafkaOutbox', topicName: $uniqueId, messageGroupId: $uniqueId),
+                    SimpleMessageChannelBuilder::createQueueChannel('orderProcessing'),
+                ]),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+    }
+}
diff --git a/packages/Redis/src/RedisBackedMessageChannelBuilder.php b/packages/Redis/src/RedisBackedMessageChannelBuilder.php
index df716069f..c0096ac6b 100644
--- a/packages/Redis/src/RedisBackedMessageChannelBuilder.php
+++ b/packages/Redis/src/RedisBackedMessageChannelBuilder.php
@@ -33,9 +33,9 @@ public static function create(string $channelName, string $connectionReferenceNa
         return new self($channelName, $connectionReferenceName);
     }
 
-    public function withAsyncPublishing(bool $asyncPublishing = true): self
+    public function withHighThroughputPublishing(bool $enabled = true): self
     {
-        $this->getRedisOutboundChannelAdapter()->withAsyncPublishing($asyncPublishing);
+        $this->getRedisOutboundChannelAdapter()->withAsyncPublishing($enabled);
 
         return $this;
     }
diff --git a/packages/Redis/src/RedisInboundChannelAdapterBuilder.php b/packages/Redis/src/RedisInboundChannelAdapterBuilder.php
index 59a88461e..524d69a58 100644
--- a/packages/Redis/src/RedisInboundChannelAdapterBuilder.php
+++ b/packages/Redis/src/RedisInboundChannelAdapterBuilder.php
@@ -40,6 +40,7 @@ public function compile(MessagingContainerBuilder $builder): Definition
             DefaultHeaderMapper::createWith($this->headerMapper, []),
             EnqueueHeader::HEADER_ACKNOWLEDGE,
             Reference::to(LoggingGateway::class),
+            $this->finalFailureStrategy,
         ]);
 
         return new Definition(RedisInboundChannelAdapter::class, [
diff --git a/packages/Redis/tests/Integration/AsyncPublishingReliabilityTest.php b/packages/Redis/tests/Integration/AsyncPublishingReliabilityTest.php
index 9c6f57af6..181039554 100644
--- a/packages/Redis/tests/Integration/AsyncPublishingReliabilityTest.php
+++ b/packages/Redis/tests/Integration/AsyncPublishingReliabilityTest.php
@@ -75,7 +75,7 @@ private function bootstrapEcotoneWithRetryingChannel(): FlowTestSupport
                     PollableChannelConfiguration::create(self::CHANNEL_NAME, RetryTemplateBuilder::fixedBackOff(1)->maxRetryAttempts(1)->build()),
                 ]),
             enableAsynchronousProcessing: [
-                RedisBackedMessageChannelBuilder::create(self::CHANNEL_NAME)->withAsyncPublishing(),
+                RedisBackedMessageChannelBuilder::create(self::CHANNEL_NAME)->withHighThroughputPublishing(),
             ],
             licenceKey: LicenceTesting::VALID_LICENCE,
         );
diff --git a/packages/Redis/tests/Integration/AsyncPublishingTest.php b/packages/Redis/tests/Integration/AsyncPublishingTest.php
index d38992d85..0116072ca 100644
--- a/packages/Redis/tests/Integration/AsyncPublishingTest.php
+++ b/packages/Redis/tests/Integration/AsyncPublishingTest.php
@@ -293,7 +293,7 @@ private function bootstrapEcotoneWithChannel(object $orderService, ?string $lice
                 ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::REDIS_PACKAGE]))
                 ->withExtensionObjects([
                     RedisBackedMessageChannelBuilder::create('asyncOrdersChannel')
-                        ->withAsyncPublishing(),
+                        ->withHighThroughputPublishing(),
                 ]),
             licenceKey: $licenceKey,
         );
diff --git a/packages/Sqs/src/SqsBackedMessageChannelBuilder.php b/packages/Sqs/src/SqsBackedMessageChannelBuilder.php
index c12326cba..cb6df98e4 100644
--- a/packages/Sqs/src/SqsBackedMessageChannelBuilder.php
+++ b/packages/Sqs/src/SqsBackedMessageChannelBuilder.php
@@ -33,9 +33,9 @@ public static function create(string $channelName, string $connectionReferenceNa
         return new self($channelName, $connectionReferenceName);
     }
 
-    public function withAsyncPublishing(bool $asyncPublishing = true, ?int $timeoutInMilliseconds = null): self
+    public function withHighThroughputPublishing(bool $enabled = true, ?int $timeoutInMilliseconds = null): self
     {
-        $this->getSqsOutboundChannelAdapter()->withAsyncPublishing($asyncPublishing, $timeoutInMilliseconds);
+        $this->getSqsOutboundChannelAdapter()->withAsyncPublishing($enabled, $timeoutInMilliseconds);
 
         return $this;
     }
diff --git a/packages/Sqs/src/SqsInboundChannelAdapterBuilder.php b/packages/Sqs/src/SqsInboundChannelAdapterBuilder.php
index 574501fbc..59c9f58af 100644
--- a/packages/Sqs/src/SqsInboundChannelAdapterBuilder.php
+++ b/packages/Sqs/src/SqsInboundChannelAdapterBuilder.php
@@ -40,6 +40,7 @@ public function compile(MessagingContainerBuilder $builder): Definition
             DefaultHeaderMapper::createWith($this->headerMapper, []),
             EnqueueHeader::HEADER_ACKNOWLEDGE,
             Reference::to(LoggingGateway::class),
+            $this->finalFailureStrategy,
         ]);
 
         return new Definition(SqsInboundChannelAdapter::class, [
diff --git a/packages/Sqs/tests/Integration/AsyncPublishingReliabilityTest.php b/packages/Sqs/tests/Integration/AsyncPublishingReliabilityTest.php
index 0e409934e..ed9707cac 100644
--- a/packages/Sqs/tests/Integration/AsyncPublishingReliabilityTest.php
+++ b/packages/Sqs/tests/Integration/AsyncPublishingReliabilityTest.php
@@ -109,7 +109,7 @@ private function bootstrapChannel(string $channelName): FlowTestSupport
                 ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::SQS_PACKAGE]))
                 ->withExtensionObjects([
                     SqsBackedMessageChannelBuilder::create($channelName)
-                        ->withAsyncPublishing(),
+                        ->withHighThroughputPublishing(),
                 ]),
             licenceKey: LicenceTesting::VALID_LICENCE,
         );
diff --git a/packages/Sqs/tests/Integration/AsyncPublishingTest.php b/packages/Sqs/tests/Integration/AsyncPublishingTest.php
index ae441c798..492c746e1 100644
--- a/packages/Sqs/tests/Integration/AsyncPublishingTest.php
+++ b/packages/Sqs/tests/Integration/AsyncPublishingTest.php
@@ -240,7 +240,7 @@ private function bootstrapEcotoneWithChannel(object $orderService, ?string $lice
                 ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::SQS_PACKAGE]))
                 ->withExtensionObjects([
                     SqsBackedMessageChannelBuilder::create('asyncOrdersChannel')
-                        ->withAsyncPublishing(),
+                        ->withHighThroughputPublishing(),
                 ]),
             licenceKey: $licenceKey,
         );