Kafka message channel

.github/workflows/split-testing.yml

@@ -115,6 +115,9 @@ jobs:
           extensions: grpc, rdkafka
           coverage: pcov
 
+      - name: Install OpenSSH (For Kafka)
+        run: sudo apt-get update && sudo apt-get install -y --no-install-recommends openssh-client
+
       - name: Enable merge-plugin
         run: composer global config --no-interaction allow-plugins.wikimedia/composer-merge-plugin true && composer global require wikimedia/composer-merge-plugin
         working-directory: ${{ matrix.package.directory }}

.github/workflows/test-monorepo.yml

@@ -87,6 +87,9 @@ jobs:
         extensions: grpc, rdkafka
         coverage: pcov
 
+    - name: Install OpenSSH (For Kafka)
+      run: sudo apt-get update && sudo apt-get install -y --no-install-recommends openssh-client
+
     - uses: actions/checkout@v2
 
     - name: Validate composer.json and composer.lock

bin/check-licence.php

@@ -13,6 +13,6 @@
     $fileContent = file_get_contents($file->getRealPath());
 
     if (! preg_match('/\*\s*(@licence|licence)\s+(Enterprise|Apache\-2\.0)\s*/', $fileContent, $matches)) {
-        throw new \RuntimeException("Missing licence in file: " . $file->getRealPath(). "\n. You can add related licence by triggering `bin/add-apache-licence.php` or if you are changing Enterprise modules `bin/add-enterprise-licence.php`");
+        throw new \RuntimeException("Missing licence in file: " . $file->getRealPath(). "\n. You can add related licence by triggering `php bin/add-apache-licence.php` or if you are changing Enterprise modules `php bin/add-enterprise-licence.php`");
     }
 }

docker-compose.yml

@@ -2,7 +2,7 @@ version: "3.8"
 
 services:
   app:
-    image: simplycodedsoftware/php:8.3.13
+    image: simplycodedsoftware/php:8.3.10
     volumes:
       - "$PWD:/data/app"
     working_dir: "/data/app"

packages/Ecotone/src/Messaging/Channel/PollableChannel/Serialization/OutboundMessageConverter.php

@@ -18,7 +18,7 @@ class OutboundMessageConverter
 {
     public function __construct(
         private HeaderMapper $headerMapper,
-        private ?MediaType $defaultConversionMediaType,
+        private ?MediaType $defaultConversionMediaType = null,
         private ?int $defaultDeliveryDelay = null,
         private ?int $defaultTimeToLive = null,
         private ?int $defaultPriority = null,

packages/Ecotone/src/Messaging/Config/Annotation/ModuleConfiguration/BasicMessagingModule.php

@@ -57,7 +57,6 @@ public static function create(AnnotationFinder $annotationRegistrationService, I
      */
     public function prepare(Configuration $messagingConfiguration, array $extensionObjects, ModuleReferenceSearchService $moduleReferenceSearchService, InterfaceToCallRegistry $interfaceToCallRegistry): void
     {
-        $serviceConfiguration = ExtensionObjectResolver::resolveUnique(ServiceConfiguration::class, $extensionObjects, ServiceConfiguration::createWithDefaults());
         foreach ($extensionObjects as $extensionObject) {
             if ($extensionObject instanceof ChannelInterceptorBuilder) {
                 $messagingConfiguration->registerChannelInterceptor($extensionObject);

packages/Enqueue/src/NullEntrypointGateway.php → packages/Ecotone/src/Messaging/Handler/Gateway/NullEntrypointGateway.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Ecotone\Enqueue;
+namespace Ecotone\Messaging\Handler\Gateway;
 
 use Ecotone\Messaging\Config\Container\DefinedObject;
 use Ecotone\Messaging\Config\Container\Definition;

packages/Enqueue/src/EnqueueAcknowledgementCallback.php

@@ -110,7 +110,7 @@ public function reject(): void
         try {
             $this->enqueueConsumer->reject($this->enqueueMessage);
         } catch (Exception $exception) {
-            $this->loggingGateway->info('Failed to reject message, disconnecting Connection in order to self-heal. Failure happen due to: ' . $exception->getMessage(), exception: $exception);
+            $this->loggingGateway->info('Failed to reject message, disconnecting Connection in order to self-heal. Failure happen due to: ' . $exception->getMessage(), ['exception' => $exception]);
 
             $this->connectionFactory->reconnect();
 
@@ -126,7 +126,7 @@ public function requeue(): void
         try {
             $this->enqueueConsumer->reject($this->enqueueMessage, true);
         } catch (Exception $exception) {
-            $this->loggingGateway->info('Failed to requeue message, disconnecting Connection in order to self-heal. Failure happen due to: ' . $exception->getMessage(), exception: $exception);
+            $this->loggingGateway->info('Failed to requeue message, disconnecting Connection in order to self-heal. Failure happen due to: ' . $exception->getMessage(), ['exception' => $exception]);
 
             $this->connectionFactory->reconnect();
 

packages/Enqueue/src/EnqueueInboundChannelAdapterBuilder.php

@@ -9,6 +9,7 @@
 use Ecotone\Messaging\Endpoint\InboundChannelAdapterEntrypoint;
 use Ecotone\Messaging\Endpoint\InterceptedChannelAdapterBuilder;
 use Ecotone\Messaging\Handler\Gateway\GatewayProxyBuilder;
+use Ecotone\Messaging\Handler\Gateway\NullEntrypointGateway;
 use Ecotone\Messaging\Handler\InterfaceToCall;
 use Ecotone\Messaging\Handler\InterfaceToCallRegistry;
 

packages/Kafka/src/Attribute/KafkaConsumer.php

@@ -6,18 +6,23 @@
 
 use Attribute;
 use Ecotone\Messaging\Attribute\MessageConsumer;
+use Ecotone\Messaging\Config\Container\DefinedObject;
+use Ecotone\Messaging\Config\Container\Definition;
+use Ecotone\Messaging\Support\Assert;
 
 /**
  * licence Enterprise
  */
 #[Attribute]
-final class KafkaConsumer extends MessageConsumer
+final class KafkaConsumer extends MessageConsumer implements DefinedObject
 {
     public function __construct(
         string $endpointId,
         private array|string $topics,
         private ?string $groupId = null
     ) {
+        Assert::notNullAndEmpty($topics, "Topics can't be empty");
+
         parent::__construct($endpointId);
     }
 
@@ -31,6 +36,18 @@ public function getTopics(): array
 
     public function getGroupId(): ?string
     {
-        return $this->groupId;
+        return $this->groupId ?? $this->getEndpointId();
+    }
+
+    public function getDefinition(): Definition
+    {
+        return new Definition(
+            self::class,
+            [
+                $this->getEndpointId(),
+                $this->topics,
+                $this->groupId
+            ]
+        );
     }
 }

packages/Kafka/src/Channel/KafkaMessageChannel.php

@@ -0,0 +1,40 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Kafka\Channel;
+
+use Ecotone\Kafka\Configuration\KafkaConsumerConfiguration;
+use Ecotone\Kafka\Inbound\KafkaInboundChannelAdapter;
+use Ecotone\Kafka\Outbound\KafkaOutboundChannelAdapter;
+use Ecotone\Messaging\Message;
+use Ecotone\Messaging\PollableChannel;
+
+/**
+ * licence Enterprise
+ */
+final class KafkaMessageChannel implements PollableChannel
+{
+    public function __construct(
+        private KafkaInboundChannelAdapter $inboundChannelAdapter,
+        private KafkaOutboundChannelAdapter $outboundChannelAdapter
+    )
+    {
+
+    }
+
+    public function send(Message $message): void
+    {
+        $this->outboundChannelAdapter->handle($message);
+    }
+
+    public function receiveWithTimeout(int $timeoutInMilliseconds): ?Message
+    {
+        return $this->inboundChannelAdapter->receiveWithTimeout($timeoutInMilliseconds);
+    }
+
+    public function receive(): ?Message
+    {
+        return $this->inboundChannelAdapter->receiveWithTimeout(KafkaConsumerConfiguration::DEFAULT_RECEIVE_TIMEOUT);
+    }
+}

packages/Kafka/src/Channel/KafkaMessageChannelBuilder.php

@@ -0,0 +1,70 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Kafka\Channel;
+
+use Ecotone\Kafka\Configuration\KafkaBrokerConfiguration;
+use Ecotone\Kafka\Configuration\KafkaConsumerConfiguration;
+use Ecotone\Kafka\Configuration\KafkaPublisherConfiguration;
+use Ecotone\Kafka\Inbound\KafkaInboundChannelAdapterBuilder;
+use Ecotone\Kafka\Outbound\KafkaOutboundChannelAdapterBuilder;
+use Ecotone\Messaging\Channel\MessageChannelBuilder;
+use Ecotone\Messaging\Config\Container\Definition;
+use Ecotone\Messaging\Config\Container\MessagingContainerBuilder;
+use Ecotone\Messaging\Config\Container\Reference;
+use Ecotone\Messaging\Conversion\MediaType;
+use Ecotone\Messaging\Handler\MessageHandlerBuilder;
+use Ramsey\Uuid\Uuid;
+
+/**
+ * licence Enterprise
+ */
+final class KafkaMessageChannelBuilder implements MessageChannelBuilder
+{
+    private function __construct(
+        private string                      $channelName,
+        public readonly string                      $topicName,
+        public readonly string                      $groupId,
+    )
+    {
+    }
+
+    public static function create(
+        string $channelName,
+        ?string $topicName = null,
+        ?string $groupId = null
+    ): self
+    {
+        return new self(
+            $channelName,
+            $topicName ?? $channelName,
+            $groupId ?? $channelName,
+        );
+    }
+
+    public function compile(MessagingContainerBuilder $builder): Definition|Reference
+    {
+        return new Definition(
+            KafkaMessageChannel::class,
+            [
+                KafkaInboundChannelAdapterBuilder::create(
+                    $this->channelName,
+                )->compile($builder),
+                KafkaOutboundChannelAdapterBuilder::create(
+                    $this->channelName,
+                )->compile($builder)
+            ]
+        );
+    }
+
+    public function getMessageChannelName(): string
+    {
+        return $this->channelName;
+    }
+
+    public function isPollable(): bool
+    {
+        return true;
+    }
+}

packages/Kafka/src/Configuration/KafkaAdmin.php

@@ -5,9 +5,13 @@
 namespace Ecotone\Kafka\Configuration;
 
 use Ecotone\Messaging\Config\ConfigurationException;
+use Ecotone\Kafka\Attribute\KafkaConsumer as KafkaConsumerAttribute;
+use Ecotone\Messaging\Support\Assert;
+use RdKafka\KafkaConsumer;
 use RdKafka\Producer;
 use RdKafka\ProducerTopic;
 use RdKafka\TopicConf;
+use RdKafka\TopicPartition;
 
 /**
  * licence Enterprise
@@ -20,27 +24,37 @@ final class KafkaAdmin
     private array $initializedProducers = [];
 
     /**
+     * @var KafkaConsumer[]
+     */
+    private array $initializedConsumers = [];
+
+    /**
+     * @param KafkaConsumerAttribute[] $kafkaConsumers
      * @param KafkaConsumerConfiguration[] $consumerConfigurations
      * @param TopicConfiguration[] $topicConfigurations
      * @param KafkaPublisherConfiguration[] $publisherConfigurations
+     * @param array<string, KafkaBrokerConfiguration> $kafkaBrokerConfigurations
      */
     public function __construct(
+        private array $kafkaConsumers,
         private array $consumerConfigurations,
         private array $topicConfigurations,
         private array $publisherConfigurations,
+        private array $kafkaBrokerConfigurations,
+        private bool $isRunningInTestMode
     ) {
 
     }
 
     public static function createEmpty(): self
     {
-        return new self([], [], []);
+        return new self([], [], [], [], [], false);
     }
 
     public function getConfigurationForConsumer(string $endpointId): KafkaConsumerConfiguration
     {
         if (! array_key_exists($endpointId, $this->consumerConfigurations)) {
-            return KafkaConsumerConfiguration::createWithDefaults($endpointId);
+            return KafkaConsumerConfiguration::createWithDefaults($endpointId)->enableKafkaDebugging();
         }
 
         return $this->consumerConfigurations[$endpointId];
@@ -55,17 +69,47 @@ public function getConfigurationForTopic(string $topicName): TopicConf
         return $this->topicConfigurations[$topicName]->getConfig();
     }
 
-    private function getConfigurationForPublisher(string $referenceName): KafkaPublisherConfiguration
+    public function getConfigurationForPublisher(string $referenceName): KafkaPublisherConfiguration
     {
         return $this->publisherConfigurations[$referenceName] ?? throw ConfigurationException::create("Publisher configuration for {$referenceName} not found");
     }
 
-    public function getProducer(string $referenceName, KafkaBrokerConfiguration $kafkaBrokerConfiguration): Producer
+    public function getConsumer(string $endpointId): KafkaConsumer
     {
-        if (! array_key_exists($referenceName, $this->initializedProducers)) {
-            $conf = $this->getConfigurationForPublisher($referenceName);
-            $conf = $conf->getAsKafkaConfig();
+        if (! array_key_exists($endpointId, $this->initializedConsumers)) {
+            Assert::keyExists($this->kafkaConsumers, $endpointId, "Consumer with endpoint id {$endpointId} not found");
+
+            $configuration = $this->getConfigurationForConsumer($endpointId);
+            $conf = $configuration->getConfig();
+            $conf->set('group.id', $this->kafkaConsumers[$endpointId]->getGroupId());
+            $kafkaBrokerConfiguration = $this->kafkaBrokerConfigurations[$configuration->getBrokerConfigurationReference()];
             $conf->set('bootstrap.servers', implode(',', $kafkaBrokerConfiguration->getBootstrapServers()));
+            $consumer = new KafkaConsumer($conf);
+
+            if ($this->isRunningForTests($kafkaBrokerConfiguration)) {
+                // ensures there is no need for repartitioning
+                $consumer->assign([new TopicPartition($this->kafkaConsumers[$endpointId]->getTopics()[0], 0)]);
+            }else {
+                $consumer->subscribe($this->kafkaConsumers[$endpointId]->getTopics());
+            }
+
+            foreach ($this->kafkaConsumers[$endpointId]->getTopics() as $topic) {
+                $consumer->subscribe([$topic]);
+            }
+            $consumer->assign([new TopicPartition($this->kafkaConsumers[$endpointId]->getTopics()[0], 0)]);
+
+            $this->initializedConsumers[$endpointId] = $consumer;
+        }
+
+        return $this->initializedConsumers[$endpointId];
+    }
+
+    public function getProducer(string $referenceName): Producer
+    {
+        if (! array_key_exists($referenceName, $this->initializedProducers)) {
+            $configuration = $this->getConfigurationForPublisher($referenceName);
+            $conf = $configuration->getAsKafkaConfig();
+            $conf->set('bootstrap.servers', implode(',', $this->kafkaBrokerConfigurations[$configuration->getBrokerConfigurationReference()]->getBootstrapServers()));
             $producer = new Producer($conf);
 
             $this->initializedProducers[$referenceName] = $producer;
@@ -74,14 +118,19 @@ public function getProducer(string $referenceName, KafkaBrokerConfiguration $kaf
         return $this->initializedProducers[$referenceName];
     }
 
-    public function getTopicForProducer(string $referenceName, KafkaBrokerConfiguration $kafkaBrokerConfiguration): ProducerTopic
+    public function getTopicForProducer(string $referenceName): ProducerTopic
     {
-        $producer = $this->getProducer($referenceName, $kafkaBrokerConfiguration);
+        $producer = $this->getProducer($referenceName);
         $topicName = $this->getConfigurationForPublisher($referenceName)->getDefaultTopicName();
 
         return $producer->newTopic(
             $topicName,
             $this->getConfigurationForTopic($topicName)
         );
     }
+
+    private function isRunningForTests(KafkaBrokerConfiguration $kafkaBrokerConfiguration): bool
+    {
+        return ($this->isRunningInTestMode && $kafkaBrokerConfiguration->isSetupForTesting() === null) || $kafkaBrokerConfiguration->isSetupForTesting() === true;
+    }
 }