Kafka integration first iteration (#403)

* Kafka integration - add package template
ecotoneframework · Oct 31, 2024 · def6e75 · def6e75
1 parent 1fce212
commit def6e75
Show file tree

Hide file tree

Showing 44 changed files with 1,616 additions and 9 deletions.
diff --git a/.github/workflows/benchmark-pr.yml b/.github/workflows/benchmark-pr.yml
@@ -58,11 +58,23 @@ jobs:
         image: redis:7-alpine
         ports:
           - '6379:6379'
+      kafka:
+        image: bitnami/kafka:latest
+        env:
+          KAFKA_CFG_NODE_ID: '0'
+          KAFKA_CFG_PROCESS_ROLES: 'controller,broker'
+          KAFKA_CFG_LISTENERS: 'PLAINTEXT://:9092,CONTROLLER://:9093'
+          KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT'
+          KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: '0@kafka:9093'
+          KAFKA_CFG_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+        ports:
+          - 9092:9092
     env:
       RABBIT_HOST: amqp://127.0.0.1:5672
       SQS_DSN: sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://127.0.0.1:4566&version=latest
       REDIS_DSN: redis://127.0.0.1:6379
       DATABASE_DSN: pgsql://ecotone:[email protected]:5432/ecotone
+      KAFKA_DSN: 127.0.0.1:9092
       APP_DB_HOST: 127.0.0.1
       APP_DB_PORT: 5432
       APP_DB_DRIVER: pdo_pgsql
@@ -71,7 +83,7 @@ jobs:
         uses: shivammathur/setup-php@v2
         with:
           php-version: ${{ matrix.php-versions }}
-          extensions: grpc
+          extensions: grpc, rdkafka
           coverage: none
 
       - uses: actions/checkout@v3

diff --git a/.github/workflows/split-testing.yml b/.github/workflows/split-testing.yml
@@ -89,18 +89,30 @@ jobs:
         image: redis:7-alpine
         ports:
           - '6379:6379'
+      kafka:
+        image: bitnami/kafka:latest
+        env:
+          KAFKA_CFG_NODE_ID: '0'
+          KAFKA_CFG_PROCESS_ROLES: 'controller,broker'
+          KAFKA_CFG_LISTENERS: 'PLAINTEXT://:9092,CONTROLLER://:9093'
+          KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT'
+          KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: '0@kafka:9093'
+          KAFKA_CFG_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+        ports:
+          - 9092:9092
     env:
       RABBIT_HOST: amqp://127.0.0.1:5672
       SQS_DSN: sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://127.0.0.1:4566&version=latest
       REDIS_DSN: redis://127.0.0.1:6379
+      KAFKA_DSN: 127.0.0.1:9092
     steps:
       - uses: actions/checkout@v2
 
       - name: Set Up PHP
         uses: shivammathur/setup-php@v2
         with:
           php-version: 8.2
-          extensions: grpc
+          extensions: grpc, rdkafka
           coverage: pcov
 
       - name: Enable merge-plugin
@@ -113,6 +125,12 @@ jobs:
         env:
           COMPOSER_ROOT_VERSION: 'dev-main'
 
+      - name: Wait for Kafka to be ready
+        run: |
+          while ! nc -z localhost 9092; do
+            sleep 1
+          done
+
       - name: Run tests on Postgres
         run: composer tests:ci
         working-directory: ${{ matrix.package.directory }}

diff --git a/.github/workflows/test-monorepo.yml b/.github/workflows/test-monorepo.yml
@@ -63,16 +63,28 @@ jobs:
         image: redis:7-alpine
         ports:
           - '6379:6379'
+      kafka:
+        image: bitnami/kafka:latest
+        env:
+          KAFKA_CFG_NODE_ID: '0'
+          KAFKA_CFG_PROCESS_ROLES: 'controller,broker'
+          KAFKA_CFG_LISTENERS: 'PLAINTEXT://:9092,CONTROLLER://:9093'
+          KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT'
+          KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: '0@kafka:9093'
+          KAFKA_CFG_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+        ports:
+          - 9092:9092
     env:
       RABBIT_HOST: amqp://127.0.0.1:5672
       SQS_DSN: sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://127.0.0.1:4566&version=latest
       REDIS_DSN: redis://127.0.0.1:6379
+      KAFKA_DSN: 127.0.0.1:9092
     steps:
     - name: PHP ${{ matrix.php-versions }} - ${{ matrix.stability }}
       uses: shivammathur/setup-php@v2
       with:
         php-version: ${{ matrix.php-versions }}
-        extensions: grpc
+        extensions: grpc, rdkafka
         coverage: pcov
 
     - uses: actions/checkout@v2
@@ -89,6 +101,12 @@ jobs:
     - name: Test PHPStan
       run: vendor/bin/phpstan
 
+    - name: Wait for Kafka to be ready
+      run: |
+        while ! nc -z localhost 9092; do
+          sleep 1
+        done
+
     - name: Test PHPUnit on Postgres
       run: vendor/bin/phpunit
       env:

diff --git a/_PackageTemplate/composer.json b/_PackageTemplate/composer.json
@@ -55,7 +55,7 @@
         },
         "merge-plugin": {
             "include": [
-                "local_packages.json"
+                "../local_packages.json"
             ]
         },
         "license-info": {

diff --git a/composer.json b/composer.json
@@ -50,6 +50,7 @@
             "Ecotone\\Messaging\\": "packages/Ecotone/src/Messaging/",
             "Ecotone\\Modelling\\": "packages/Ecotone/src/Modelling/",
             "Ecotone\\OpenTelemetry\\": "packages/OpenTelemetry/src",
+            "Ecotone\\Kafka\\": "packages/Kafka/src",
             "Ecotone\\SymfonyBundle\\": "packages/Symfony/SymfonyBundle",
             "Ecotone\\SymfonyBundle\\DependencyInjection\\": "packages/Symfony/DependencyInjection",
             "Monorepo\\": "Monorepo"
@@ -96,6 +97,9 @@
             "Test\\Ecotone\\OpenTelemetry\\": [
                 "packages/OpenTelemetry/tests"
             ],
+            "Test\\Ecotone\\Kafka\\": [
+                "packages/Kafka/tests"
+            ],
             "Test\\Ecotone\\Laravel\\": "packages/Laravel/tests",
             "App\\MultiTenant\\": "packages/Laravel/tests/MultiTenant/app",
             "App\\Licence\\Laravel\\": "packages/Laravel/tests/Licence/app",
@@ -160,7 +164,8 @@
         "nesbot/carbon": "^2.71",
         "moneyphp/money": "^4.1.0",
         "timacdonald/log-fake": "^2.0",
-        "symfony/monolog-bundle": "^3.10"
+        "symfony/monolog-bundle": "^3.10",
+        "kwn/php-rdkafka-stubs": "^2.2"
     },
     "conflict": {
         "symfony/doctrine-messenger": ">7.0.5 < 7.1.0"

diff --git a/docker-compose.yml b/docker-compose.yml
@@ -2,7 +2,7 @@ version: "3.8"
 
 services:
   app:
-    image: simplycodedsoftware/php:8.2.12
+    image: simplycodedsoftware/php:8.3.10
     volumes:
       - "$PWD:/data/app"
     working_dir: "/data/app"
@@ -25,11 +25,12 @@ services:
       DATABASE_MYSQL: mysql://ecotone:secret@database-mysql:3306/ecotone
       SQS_DSN: sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
       REDIS_DSN: redis://redis:6379
+      KAFKA_DSN: kafka:9092
       APP_MERGE_PLUGIN: "yes"
     env_file:
       - ".env"
   app8_1:
-    image: simplycodedsoftware/php:8.1.25
+    image: simplycodedsoftware/php:8.1.29
     volumes:
       - "$PWD:/data/app"
     working_dir: "/data/app"
@@ -52,11 +53,12 @@ services:
       DATABASE_MYSQL: mysql://ecotone:secret@database-mysql:3306/ecotone
       SQS_DSN: sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
       REDIS_DSN: redis://redis:6379
+      KAFKA_DSN: kafka:9092
       APP_MERGE_PLUGIN: "yes"
     env_file:
       - ".env"
   app8_0:
-    image: simplycodedsoftware/php:8.0
+    image: simplycodedsoftware/nginx-php:8.0.30
     volumes:
       - "$PWD:/data/app"
     working_dir: "/data/app"
@@ -79,6 +81,7 @@ services:
       DATABASE_MYSQL: mysql://ecotone:secret@database-mysql:3306/ecotone
       SQS_DSN: sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
       REDIS_DSN: redis://redis:6379
+      KAFKA_DSN: kafka:9092
       APP_MERGE_PLUGIN: "yes"
     env_file:
       - ".env"
@@ -143,3 +146,24 @@ services:
       - default
     ports:
       - 16686:16686
+  kafka:
+    image: 'bitnami/kafka:latest'
+    ports:
+      - '9092:9092'
+    environment:
+      - KAFKA_CFG_NODE_ID=0
+      - KAFKA_CFG_PROCESS_ROLES=controller,broker
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
+      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+    networks:
+      - default
+  kafdrop:
+    image: 'obsidiandynamics/kafdrop:latest'
+    ports:
+      - '9999:9000'
+    environment:
+      - KAFKA_BROKERCONNECT=kafka:9092
+    networks:
+      - default
diff --git a/local-packages.json b/local-packages.json
@@ -35,6 +35,10 @@
       {
         "type": "path",
         "url": "packages/LiteApplication"
+      },
+      {
+        "type": "path",
+        "url": "packages/Kafka"
       }
     ]
 }
diff --git a/packages/Ecotone/src/Messaging/Config/Container/Definition.php b/packages/Ecotone/src/Messaging/Config/Container/Definition.php
@@ -19,6 +19,14 @@ public function __construct(protected string $className, protected array $argume
     {
     }
 
+    /**
+     * @param array<string|int, mixed> $arguments
+     */
+    public static function createFor(string $className, array $arguments): self
+    {
+        return new self($className, $arguments);
+    }
+
     public function getClassName(): string
     {
         return $this->className;

diff --git a/packages/Ecotone/src/Messaging/Config/ModuleClassList.php b/packages/Ecotone/src/Messaging/Config/ModuleClassList.php
@@ -17,6 +17,7 @@
 use Ecotone\EventSourcing\Config\EventSourcingModule;
 use Ecotone\JMSConverter\Configuration\JMSConverterConfigurationModule;
 use Ecotone\JMSConverter\Configuration\JMSDefaultSerialization;
+use Ecotone\Kafka\Configuration\KafkaModule;
 use Ecotone\Laravel\Config\LaravelConnectionModule;
 use Ecotone\Lite\Test\Configuration\EcotoneTestSupportModule;
 use Ecotone\Messaging\Channel\Collector\Config\CollectorModule;
@@ -157,4 +158,8 @@ class ModuleClassList
     public const SYMFONY_MODULES = [
         SymfonyConnectionModule::class,
     ];
+
+    public const KAFKA_MODULES = [
+        KafkaModule::class,
+    ];
 }
diff --git a/packages/Ecotone/src/Messaging/Config/ModulePackageList.php b/packages/Ecotone/src/Messaging/Config/ModulePackageList.php
@@ -15,6 +15,7 @@ final class ModulePackageList
     public const DBAL_PACKAGE = 'dbal';
     public const REDIS_PACKAGE = 'redis';
     public const SQS_PACKAGE = 'sqs';
+    public const KAFKA_PACKAGE = 'kafka';
     public const EVENT_SOURCING_PACKAGE = 'eventSourcing';
     public const JMS_CONVERTER_PACKAGE = 'jmsConverter';
     public const TRACING_PACKAGE = 'tracing';
@@ -31,6 +32,7 @@ public static function getModuleClassesForPackage(string $packageName): array
             ModulePackageList::DBAL_PACKAGE => ModuleClassList::DBAL_MODULES,
             ModulePackageList::REDIS_PACKAGE => ModuleClassList::REDIS_MODULES,
             ModulePackageList::SQS_PACKAGE => ModuleClassList::SQS_MODULES,
+            ModulePackageList::KAFKA_PACKAGE => ModuleClassList::KAFKA_MODULES,
             ModulePackageList::EVENT_SOURCING_PACKAGE => ModuleClassList::EVENT_SOURCING_MODULES,
             ModulePackageList::JMS_CONVERTER_PACKAGE => ModuleClassList::JMS_CONVERTER_MODULES,
             ModulePackageList::TRACING_PACKAGE => ModuleClassList::TRACING_MODULES,
@@ -52,6 +54,7 @@ public static function allPackages(): array
             self::AMQP_PACKAGE,
             self::REDIS_PACKAGE,
             self::SQS_PACKAGE,
+            self::KAFKA_PACKAGE,
             self::DBAL_PACKAGE,
             self::EVENT_SOURCING_PACKAGE,
             self::JMS_CONVERTER_PACKAGE,

diff --git a/packages/Enqueue/src/InboundMessageConverter.php b/packages/Enqueue/src/InboundMessageConverter.php
@@ -57,6 +57,11 @@ public function toMessage(EnqueueMessage $source, EnqueueConsumer $consumer, Con
                 ->setHeader(MessageHeaders::MESSAGE_CORRELATION_ID, $enqueueMessageHeaders[MessageHeaders::MESSAGE_CORRELATION_ID]);
         }
 
+        if (isset($enqueueMessageHeaders[MessageHeaders::PARENT_MESSAGE_ID])) {
+            $messageBuilder = $messageBuilder
+                ->setHeader(MessageHeaders::PARENT_MESSAGE_ID, $enqueueMessageHeaders[MessageHeaders::PARENT_MESSAGE_ID]);
+        }
+
         return $messageBuilder
             ->setHeader(MessageHeaders::CONSUMER_ACK_HEADER_LOCATION, $this->acknowledgeHeaderName);
     }

diff --git a/packages/Kafka/.gitattributes b/packages/Kafka/.gitattributes
@@ -0,0 +1,7 @@
+tests/          export-ignore
+.coveralls.yml  export-ignore
+.gitattributes  export-ignore
+.gitignore      export-ignore
+behat.yaml      export-ignore
+phpstan.neon    export-ignore
+phpunit.xml     export-ignore
diff --git a/packages/Kafka/.github/FUNDING.yml b/packages/Kafka/.github/FUNDING.yml
@@ -0,0 +1,12 @@
+# These are supported funding model platforms
+
+github: [dgafka]
+patreon: # Replace with a single Open Collective username
+open_collective: # Replace with a single Open Collective username
+ko_fi: # Replace with a single Ko-fi username
+tidelift: # Replace with a single Tidelift platform-name/package-name e.g., npm/babel
+community_bridge: # Replace with a single Community Bridge project-name e.g., cloud-foundry
+liberapay: # Replace with a single Liberapay username
+issuehunt: # Replace with a single IssueHunt username
+otechie: # Replace with a single Otechie username
+custom: # Replace with up to 4 custom sponsorship URLs e.g., ['link1', 'link2']
diff --git a/packages/Kafka/.github/ISSUE_TEMPLATE/bug_report.md b/packages/Kafka/.github/ISSUE_TEMPLATE/bug_report.md
@@ -0,0 +1,10 @@
+---
+name: This is Read-Only repository
+about: Report at ecotoneframework/ecotone-dev
+title: ''
+labels: ''
+assignees: ''
+
+---
+
+Report issue at [ecotone-dev](ecotoneframework/ecotone-dev)
diff --git a/packages/Kafka/.gitignore b/packages/Kafka/.gitignore
@@ -0,0 +1,9 @@
+.idea/
+vendor/
+bin/
+tests/coverage
+!tests/coverage/.gitkeep
+file
+.phpunit.result.cache
+composer.lock
+phpunit.xml
diff --git a/packages/Kafka/LICENSE b/packages/Kafka/LICENSE
@@ -0,0 +1,4 @@
+Copyright (c) 2024 Dariusz Gafka <[email protected]>
+
+Licensed under the Ecotone Enterprise License.
+For details on the Enterprise License, please write to [email protected].