Skip to content

Add KafkaDriver for Apache Kafka through ext-rdkafka - #155

Merged
tomaj merged 2 commits into
masterfrom
feat/kafka-driver
Oct 23, 2026
Merged

Add KafkaDriver for Apache Kafka through ext-rdkafka#155
tomaj merged 2 commits into
masterfrom
feat/kafka-driver

Conversation

@tomaj

@tomaj tomaj commented Oct 19, 2026 •
edited
Loading

Copy link
Copy Markdown
Owner

Closes #14

Summary

Adds KafkaDriver for Apache Kafka through ext-rdkafka 6 (Composer suggest). send() produces to one topic keyed by the message id and waits for confirmation; wait() consumes in a consumer group and commits the offset only after the handler, so delivery is at least once. A payload that cannot be unserialized is committed and reported so it cannot block a partition. No priorities and no delay (same shape as RabbitMQ and SQS).

Review (merge-ready)

  • DriverInterface parity with RabbitMQ/SQS: setupPriorityQueue() → NotSupportedException (message aligned with SQS/Rabbit … now); send() / wait() ignore priority args (documented in Limits).
  • Commit behaviour: after callback; SerializeException commits then rethrows (no partition poison); ShutdownException commits then rethrows; other exceptions leave offset uncommitted.
  • Infection on KafkaDriver.php + KafkaDriverException.php: 51/51 killed, Covered MSI 100%.
  • Docs: six drivers, FAQ, diagrams, examples/kafka 1:1 including bin/shutdown.php + SharedFileShutdown.
  • CHANGELOG / UPGRADE / composer suggest / PHP 8.4+8.5 / security: OK.
  • 1 commit Tomas Majer <tomasmajer@gmail.com>.

CI workflows — paste again (required)

The bot cannot push .github/workflows/* (GitHub OAuth lacks workflow scope). You already pasted them once (e5ce45f); a later force-push of the review fixes dropped that commit. Please paste the two files below again (same content as before).

File On master After paste
.github/workflows/ci.yml Redis, RabbitMQ, SQS + kafka service, rdkafka, HERMES_TEST_KAFKA, Require ext-rdkafka
.github/workflows/examples.yml recipes + frameworks + filesystem in matrix + dedicated kafka job

Until then PHP 8.4 coverage stays red (~Kafka unit tests skip without rdkafka).

1. Replace .github/workflows/ci.yml entirely

name: CI

on:
  push:
    branches: [ master, main ]
  pull_request:
    branches: [ master, main ]

permissions:
  contents: read

jobs:
  tests:
    name: PHP ${{ matrix.php }} ${{ matrix.dependency-version }}
    runs-on: ubuntu-latest
    permissions:
      contents: read
      actions: write

    strategy:
      fail-fast: false
      matrix:
        php: ['8.4', '8.5']
        dependency-version: ['']
        include:
          - php: '8.4'
            dependency-version: '--prefer-lowest'

    services:
      redis:
        image: redis:8.8.3-alpine@sha256:0b2b77d3ea5078274795e3177cdbdada8b96316684a38911d528534ed679b5ec
        ports: ['6379:6379']
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 2s
          --health-timeout 5s
          --health-retries 30
      rabbitmq:
        image: rabbitmq:4.3.6-alpine@sha256:6931816af1a943f471c038b278835d36692d55dec769232739296bb91b7de0f5
        ports: ['5672:5672']
        options: >-
          --health-cmd "rabbitmq-diagnostics -q ping"
          --health-interval 5s
          --health-timeout 10s
          --health-retries 30
      sqs:
        image: motoserver/moto:5.2.4@sha256:120b1589c6df7e35d2092935f6ff724486bb2c4189d0e36ce795baba1400ffd2
        ports: ['4566:5000']
      kafka:
        image: apache/kafka:4.3.1@sha256:77e3df9054047a88b520d0cc46e16696d3b22022e1d580aeccd2632df6532837
        ports: ['9092:9092']
        env:
          KAFKA_NODE_ID: 1
          KAFKA_PROCESS_ROLES: broker,controller
          KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092
          KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
          KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
          KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
          KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
          KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
        options: >-
          --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server 127.0.0.1:9092"
          --health-interval 5s
          --health-timeout 20s
          --health-retries 30

    env:
      HERMES_TEST_REDIS: tcp://127.0.0.1:6379
      HERMES_TEST_RABBITMQ: 127.0.0.1
      HERMES_TEST_SQS: http://127.0.0.1:4566
      HERMES_TEST_KAFKA: 127.0.0.1:9092
      HERMES_REQUIRE_SERVICES: '1'

    steps:
    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

    - name: Setup PHP
      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
      with:
        php-version: ${{ matrix.php }}
        extensions: mbstring, xml, redis, rdkafka
        coverage: xdebug

    - name: Validate composer.json
      run: composer validate --strict

    - name: Cache Composer packages
      uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
      with:
        path: vendor
        key: ${{ runner.os }}-php-${{ matrix.php }}-${{ matrix.dependency-version }}-${{ hashFiles('**/composer.lock') }}
        restore-keys: |
          ${{ runner.os }}-php-${{ matrix.php }}-${{ matrix.dependency-version }}-

    - name: Install dependencies
      run: composer update ${{ matrix.dependency-version }} --prefer-dist --no-progress --no-interaction

    - name: Require ext-rdkafka
      # the Kafka tests skip without the extension; fail instead
      run: php --ri rdkafka

    - name: Wait for moto
      run: curl --silent --show-error --fail --retry 30 --retry-delay 1 --retry-all-errors --output /dev/null http://127.0.0.1:4566/moto-api/

    - name: Run tests
      run: XDEBUG_MODE=coverage vendor/bin/phpunit --coverage-clover build/logs/clover.xml --coverage-html build/coverage

    - name: Require 95% statement coverage
      if: matrix.php == '8.4' && matrix.dependency-version == ''
      run: |
        php -r '
          $xml = simplexml_load_file("build/logs/clover.xml");
          $metrics = $xml->project->metrics;
          $statements = (int) $metrics["statements"];
          $covered = (int) $metrics["coveredstatements"];
          $percent = $statements === 0 ? 0 : ($covered / $statements) * 100;
          echo sprintf("Statements: %d/%d (%.2f%%)\n", $covered, $statements, $percent);
          if ($percent < 95) {
              fwrite(STDERR, "Statement coverage is below 95%.\n");
              exit(1);
          }
        '

    - name: Upload coverage to Coveralls
      if: matrix.php == '8.4' && matrix.dependency-version == ''
      uses: coverallsapp/github-action@8d6379e14d29928660c4ba802d8e85393440b329 # v2.3.8
      with:
        github-token: ${{ secrets.GITHUB_TOKEN }}
        file: build/logs/clover.xml
        format: clover
        fail-on-error: false

    - name: Upload coverage reports
      uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2
      if: github.event_name == 'push' && github.ref == 'refs/heads/master' && matrix.php == '8.4' && matrix.dependency-version == ''
      with:
        name: coverage-report
        path: build/coverage

  code-style:
    name: Code Style (PER-CS/PSR-12 + PHP 8.4)
    runs-on: ubuntu-latest
    permissions:
      contents: read

    steps:
    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

    - name: Setup PHP
      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
      with:
        php-version: '8.4'
        extensions: mbstring, xml, redis

    - name: Install dependencies
      run: composer update --prefer-dist --no-progress --no-interaction

    - name: Check code style
      run: vendor/bin/php-cs-fixer fix --dry-run --diff --verbose

  static-analysis:
    name: Static Analysis (PHPStan level max)
    runs-on: ubuntu-latest
    permissions:
      contents: read

    steps:
    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

    - name: Setup PHP
      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
      with:
        php-version: '8.4'
        extensions: mbstring, xml, redis

    - name: Install dependencies
      run: composer update --prefer-dist --no-progress --no-interaction

    - name: Run PHPStan
      run: vendor/bin/phpstan analyse --memory-limit=1G

  rector:
    name: Rector
    runs-on: ubuntu-latest
    permissions:
      contents: read

    steps:
    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

    - name: Setup PHP
      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
      with:
        php-version: '8.4'
        extensions: mbstring, xml, redis

    - name: Install dependencies
      run: composer update --prefer-dist --no-progress --no-interaction

    - name: Run Rector
      run: vendor/bin/rector --dry-run --no-progress-bar

  security:
    name: Security Audit
    runs-on: ubuntu-latest
    permissions:
      contents: read

    steps:
    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

    - name: Setup PHP
      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
      with:
        php-version: '8.4'
        extensions: redis

    - name: Install dependencies
      run: composer update --prefer-dist --no-progress --no-interaction

    - name: Run security audit
      run: composer audit

  mutation-testing:
    name: Mutation Testing (Infection)
    runs-on: ubuntu-latest
    permissions:
      contents: read

    services:
      redis:
        image: redis:8.8.3-alpine@sha256:0b2b77d3ea5078274795e3177cdbdada8b96316684a38911d528534ed679b5ec
        ports: ['6379:6379']
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 2s
          --health-timeout 5s
          --health-retries 30
      rabbitmq:
        image: rabbitmq:4.3.6-alpine@sha256:6931816af1a943f471c038b278835d36692d55dec769232739296bb91b7de0f5
        ports: ['5672:5672']
        options: >-
          --health-cmd "rabbitmq-diagnostics -q ping"
          --health-interval 5s
          --health-timeout 10s
          --health-retries 30
      sqs:
        image: motoserver/moto:5.2.4@sha256:120b1589c6df7e35d2092935f6ff724486bb2c4189d0e36ce795baba1400ffd2
        ports: ['4566:5000']
      kafka:
        image: apache/kafka:4.3.1@sha256:77e3df9054047a88b520d0cc46e16696d3b22022e1d580aeccd2632df6532837
        ports: ['9092:9092']
        env:
          KAFKA_NODE_ID: 1
          KAFKA_PROCESS_ROLES: broker,controller
          KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092
          KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
          KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
          KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
          KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
          KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
        options: >-
          --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server 127.0.0.1:9092"
          --health-interval 5s
          --health-timeout 20s
          --health-retries 30

    env:
      HERMES_TEST_REDIS: tcp://127.0.0.1:6379
      HERMES_TEST_RABBITMQ: 127.0.0.1
      HERMES_TEST_SQS: http://127.0.0.1:4566
      HERMES_TEST_KAFKA: 127.0.0.1:9092
      HERMES_REQUIRE_SERVICES: '1'

    steps:
    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

    - name: Setup PHP
      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
      with:
        php-version: '8.4'
        extensions: mbstring, xml, redis, rdkafka
        coverage: xdebug

    - name: Install dependencies
      run: composer update --prefer-dist --no-progress --no-interaction

    - name: Require ext-rdkafka
      # the Kafka tests skip without the extension; fail instead
      run: php --ri rdkafka

    - name: Wait for moto
      run: curl --silent --show-error --fail --retry 30 --retry-delay 1 --retry-all-errors --output /dev/null http://127.0.0.1:4566/moto-api/

    - name: Run Infection
      run: XDEBUG_MODE=coverage vendor/bin/infection --threads=1 --logger-github

2. Replace .github/workflows/examples.yml entirely

Adds filesystem to the recipes matrix and a dedicated kafka job (ext-rdkafka + Kafka 4.3.1 service, KAFKA_BROKERS).

name: Examples

# Runs the e2e.sh of every recipe in examples/ and of the three framework integrations against the
# code of the push or pull request (a Composer path repository, see examples/ci/use-local-hermes.sh).
on:
  push:
    branches: [ master, main ]
  pull_request:
    branches: [ master, main ]

permissions:
  contents: read

# a new push to a pull request cancels its older run
concurrency:
  group: examples-${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
  cancel-in-progress: ${{ github.event_name == 'pull_request' }}

jobs:
  recipes:
    name: Example ${{ matrix.example }}
    runs-on: ubuntu-latest
    timeout-minutes: 15
    strategy:
      fail-fast: false
      matrix:
        example: [welcome-email, webhooks, reminder, csv-import, invoice-pdf, priority, thumbnails, filesystem]

    services:
      redis:
        image: redis:8.8.3-alpine@sha256:0b2b77d3ea5078274795e3177cdbdada8b96316684a38911d528534ed679b5ec
        ports: ['6379:6379']
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 2s
          --health-timeout 5s
          --health-retries 30
      rabbitmq:
        image: rabbitmq:4.3.6-alpine@sha256:6931816af1a943f471c038b278835d36692d55dec769232739296bb91b7de0f5
        ports: ['5672:5672']
        options: >-
          --health-cmd "rabbitmq-diagnostics -q ping"
          --health-interval 5s
          --health-timeout 10s
          --health-retries 30
      mailpit:
        image: axllent/mailpit:v1.31.6@sha256:23fe39e81bde596a918000b3135c0f85dc8bb4c9ea6be274b051143f3e300387
        ports: ['1025:1025', '8025:8025']

    steps:
    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      with:
        persist-credentials: false

    - name: Setup PHP
      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
      with:
        php-version: '8.4'
        extensions: mbstring, xml, gd, redis
        coverage: none

    - name: Run examples/${{ matrix.example }}/e2e.sh
      env:
        EXAMPLE: ${{ matrix.example }}
      run: examples/ci/recipe.sh "$EXAMPLE"

  frameworks:
    name: Framework ${{ matrix.framework }}
    runs-on: ubuntu-latest
    timeout-minutes: 20
    strategy:
      fail-fast: false
      matrix:
        framework: [laravel, symfony, nette]

    services:
      redis:
        image: redis:8.8.3-alpine@sha256:0b2b77d3ea5078274795e3177cdbdada8b96316684a38911d528534ed679b5ec
        ports: ['6379:6379']
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 2s
          --health-timeout 5s
          --health-retries 30
      rabbitmq:
        image: rabbitmq:4.3.6-alpine@sha256:6931816af1a943f471c038b278835d36692d55dec769232739296bb91b7de0f5
        ports: ['5672:5672']
        options: >-
          --health-cmd "rabbitmq-diagnostics -q ping"
          --health-interval 5s
          --health-timeout 10s
          --health-retries 30
      sqs:
        image: motoserver/moto:5.2.4@sha256:120b1589c6df7e35d2092935f6ff724486bb2c4189d0e36ce795baba1400ffd2
        ports: ['4566:5000']

    env:
      HERMES_SQS_ENDPOINT: http://127.0.0.1:4566
      AWS_DEFAULT_REGION: eu-central-1
      AWS_ACCESS_KEY_ID: test
      AWS_SECRET_ACCESS_KEY: test

    steps:
    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      with:
        persist-credentials: false

    - name: Setup PHP
      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
      with:
        php-version: '8.4'
        extensions: mbstring, xml, redis, sqlite3, pdo_sqlite
        coverage: none

    - name: Wait for moto
      run: curl --silent --show-error --fail --retry 30 --retry-delay 1 --retry-all-errors --output /dev/null http://127.0.0.1:4566/moto-api/

    - name: Run examples/frameworks/${{ matrix.framework }}/e2e.sh
      env:
        FRAMEWORK: ${{ matrix.framework }}
      run: examples/ci/framework.sh "$FRAMEWORK" redis predis rabbitmq sqs

  kafka:
    name: Example kafka
    runs-on: ubuntu-latest
    timeout-minutes: 20
    services:
      kafka:
        image: apache/kafka:4.3.1@sha256:77e3df9054047a88b520d0cc46e16696d3b22022e1d580aeccd2632df6532837
        ports: ['9092:9092']
        env:
          KAFKA_NODE_ID: 1
          KAFKA_PROCESS_ROLES: broker,controller
          KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092
          KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
          KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
          KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
          KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
          KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
        options: >-
          --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server 127.0.0.1:9092"
          --health-interval 5s
          --health-timeout 20s
          --health-retries 30

    steps:
    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      with:
        persist-credentials: false

    - name: Setup PHP
      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
      with:
        php-version: '8.4'
        extensions: mbstring, xml, rdkafka
        coverage: none

    - name: Require ext-rdkafka
      run: php --ri rdkafka

    - name: Run examples/kafka/e2e.sh
      env:
        EXAMPLE: kafka
        KAFKA_BROKERS: 127.0.0.1:9092
      run: examples/ci/recipe.sh kafka

After paste

  1. Wait for Actions green (PHP 8.4/8.5/prefer-lowest, Infection, Examples incl. filesystem + kafka, docs, etc.).
  2. Merge when you are ready (bot will not merge).

@tomaj tomaj added this to the 6.0.0 milestone Oct 19, 2026
@tomaj tomaj self-assigned this Oct 19, 2026
This was referenced Oct 22, 2026
Merged
Merged
@tomaj
tomaj force-pushed the feat/kafka-driver branch 2 times, most recently from 2d4a201 to f04443c Compare October 23, 2026 11:27
@tomaj
Add KafkaDriver for Apache Kafka through ext-rdkafka
f90b28b 
send() produces to one topic keyed by the message id and waits for the
confirmation; wait() consumes in a consumer group and commits the offset
after the handler, so delivery is at least once. A message that cannot be
unserialized is committed and reported so it cannot block its partition.
No priorities and no delay, like RabbitMQ and SQS.

ext-rdkafka is a suggest. Integration tests run against Kafka 4.3.1
(compose.yaml, HERMES_TEST_KAFKA). Docs page with diagrams and FAQ,
examples/kafka with an e2e check.

Closes #14
@coveralls

coveralls commented Oct 23, 2026 •
edited
Loading

Copy link
Copy Markdown

Coverage Report for CI Build 39961269442

Coverage increased (+0.08%) to 98.78%

Details

  • Coverage increased (+0.08%) from the base build.
  • Patch coverage: 39 of 39 lines across 1 file are fully covered (100%).
  • No coverage regressions found.

Uncovered Changes

No uncovered changes found.

Coverage Regressions

No coverage regressions found.

Coverage Stats

Coverage Status
Relevant Lines: 656
Covered Lines: 648
Line Coverage: 98.78%
Coverage Strength: 22.27 hits per line
💛 - Coveralls

@tomaj
tomaj force-pushed the feat/kafka-driver branch from e5ce45f to f90b28b Compare October 23, 2026 13:19
@tomaj
tomaj merged commit 8f4a5c5 into master Oct 23, 2026
27 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

@tomaj tomaj

Labels

None yet

Projects

None yet

Milestone

6.0.0

Development

Successfully merging this pull request may close these issues.

Kafka Driver

2 participants

@tomaj @coveralls