README

A ready-to-use PHP library for seamless communication with Rossel's Kafka infrastructure, handling both production and consumption of messages.

Installation

composer require rossel/rossel-kafka

Upgrading? Read UPGRADE-0.5.md (from 0.4) and UPGRADE-0.7.md (from 0.6).

Configuration

rossel_kafka:
    broker:
        url: '%env(ROSSEL_KAFKA_BROKER_URL)%'
        topics:
            account_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_ACCOUNT_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            authentication_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_AUTHENTICATION_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            public_dead_letter_inout_v1_json_delete_d30: '%env(KAFKA_TOPIC_PUBLIC_DEAD_LETTER_INOUT_V1_JSON_DELETE_D30)%'
            erp_subscription_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_ERP_SUBSCRIPTION_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            public_inheritance_output_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_INHERITANCE_OUTPUT_V1_JSON_DELETE)%'
            inheritance_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_INHERITANCE_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            notification_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_NOTIFICATION_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            public_notification_input_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_NOTIFICATION_INPUT_V1_JSON_DELETE)%'
            offer_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_OFFER_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            public_offer_output_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_OFFER_OUTPUT_V1_JSON_DELETE)%'
            profile_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_PROFILE_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            purchase_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_PURCHASE_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            public_offer_input_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_OFFER_INPUT_V1_JSON_DELETE)%'
            public_subscription_input_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_SUBSCRIPTION_INPUT_V1_JSON_DELETE)%'
            public_subscription_output_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_SUBSCRIPTION_OUTPUT_V1_JSON_DELETE)%'
            public_contact_input_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_CONTACT_INPUT_V1_JSON_DELETE)%'
            input_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_INPUT_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            output_api_public_log_output_v1_json_delete: '%env(KAFKA_TOPIC_OUTPUT_API_PUBLIC_LOG_OUTPUT_V1_JSON_DELETE)%'
            public_profile_output_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_PROFILE_OUTPUT_V1_JSON_DELETE)%'
            public_b2b_customer_output_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_B2B_CUSTOMER_OUTPUT_V1_JSON_DELETE)%'
            public_b2b_sales_rep_output_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_B2B_SALES_REP_OUTPUT_V1_JSON_DELETE)%'
            public_b2b_reference_item_output_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_B2B_REFERENCE_ITEM_OUTPUT_V1_JSON_DELETE)%'
    producer:
        app_name: '%env(ROSSEL_KAFKA_PRODUCER_APP_NAME)%'
        # Send an EXEC_SUCCESS/EXEC_ERROR log message after each consumed message (default: true)
        send_log_messages: true

Each topic key maps to the actual Kafka topic name provided via environment variable.

Topic roles

By default, the direction of a topic is derived from its name, from the point of view of an application that does not own the topic:

Topic	Direction
`input`	produce
`output`	consume
`inout`	both

This default is wrong for the topic owner, and for any application granted other roles in the owner's manifest. In those cases, declare the roles granted to the application on the topic, as in the manifest ACLs:

rossel_kafka:
    broker:
        topics:
            # Short form: direction derived from the name
            public_offer_output_v1_json_delete: '%env(KAFKA_TOPIC_PUBLIC_OFFER_OUTPUT_V1_JSON_DELETE)%'
            # Long form: direction resolved from the roles
            public_offer_input_v1_json_delete:
                name: '%env(KAFKA_TOPIC_PUBLIC_OFFER_INPUT_V1_JSON_DELETE)%'
                roles: [consumer]

Accepted roles are producer and consumer. Both roles give a topic that can be produced to and consumed. Without roles (or with an empty list), the direction is derived from the name.

Producing to a topic that cannot be produced to, or consuming a topic that cannot be consumed, throws an UnauthorizedTopicOperationException.

Broker authentication

Authentication is optional — typically required in production, but not in local development.

All authentication options live under broker.authentication in the bundle config. src/Resources/config/packages/rossel_kafka.yaml wires every option to an env var: copy it to your app's config/packages/.

SASL

Set both username and password to enable SASL:

ROSSEL_KAFKA_BROKER_SASL_USERNAME=my-user
ROSSEL_KAFKA_BROKER_SASL_PASSWORD=my-password
# Mechanism: PLAIN (default), SCRAM-SHA-256 or SCRAM-SHA-512
ROSSEL_KAFKA_BROKER_SASL_MECHANISM=PLAIN

SSL — CA certificate

Two options, mutually exclusive — the local file takes priority:

From a local file:

ROSSEL_KAFKA_BROKER_SSL_CA_CERTIFICATE_PATH=/path/to/ca.pem

From a URL (downloaded automatically, cached in /tmp):

ROSSEL_KAFKA_BROKER_SSL_CA_CERTIFICATE_URL=https://...

mTLS — client certificate

For mutual TLS, provide the client certificate and private key in addition to the CA certificate above. Each variable accepts either a PEM file path or raw PEM content:

ROSSEL_KAFKA_BROKER_SSL_CLIENT_CERTIFICATE=/path/to/client.crt.pem
ROSSEL_KAFKA_BROKER_SSL_CLIENT_KEY=/path/to/client.key.pem
# Optional: password protecting the private key
ROSSEL_KAFKA_BROKER_SSL_CLIENT_KEY_PASSWORD=secret

Security protocol — auto-selected

The bundle selects the security.protocol rdkafka option automatically based on which credentials are provided:

CA cert / client cert	SASL	Protocol
no	no	`plaintext`
yes	no	`ssl`
no	yes	`sasl_plaintext`
yes	yes	`sasl_ssl`

Unset or empty variables are treated as null — no configuration needed for unauthenticated brokers.

Usage

Send a message to a topic

use Rossel\RosselKafka\Enum\Config\Broker\TopicConfigKeys;
use Rossel\RosselKafka\Enum\MessageHeaders\Area;
use Rossel\RosselKafka\Enum\MessageHeaders\MessageType;
use Rossel\RosselKafka\Model\Message;
use Rossel\RosselKafka\Model\MessageHeaders;
use Rossel\RosselKafka\Service\Connector\KafkaConnector;
use Rossel\RosselKafka\Service\KafkaTopicsFetcher;

// Via Symfony DI (recommended)
$topic = $kafkaTopicsFetcher->get(TopicConfigKeys::KAFKA_TOPIC_PUBLIC_SUBSCRIPTION_INPUT_V1_JSON_DELETE);

$message = new Message(
    headers: new MessageHeaders(
        area: Area::FRANCE,
        from: 'my-app',
        messageType: MessageType::CANCEL_B2C_SUBSCRIPTION,
    ),
    body: ['foo' => 'bar'],
    // Optional Kafka record key. Records with the same key go to the same partition,
    // so their order is kept. Without a key, records are spread across partitions.
    key: 'subscription-42',
);

$kafkaConnector->send($topic, $message);

Consume messages

Implement ConsumerInterface. With autoconfiguration (the Symfony default), the service is tagged rossel_kafka.consumer for you:

use Rossel\RosselKafka\Consumer\ConsumerInterface;
use Rossel\RosselKafka\Enum\MessageHeaders\MessageType;
use Rossel\RosselKafka\Model\Message;
use Rossel\RosselKafka\Model\Topic;

final class MyConsumer implements ConsumerInterface
{
    public function supportsTopic(Topic $topic): bool
    {
        return TopicConfigKeys::KAFKA_TOPIC_PUBLIC_SUBSCRIPTION_INPUT_V1_JSON_DELETE === $topic->getConfigKey();
    }

    public function supportsMessageType(Message $message): bool
    {
        return MessageType::CANCEL_B2C_SUBSCRIPTION === $message->getType();
    }

    public function __invoke(Message $message): void
    {
        // handle message
        // $message->getKey() returns the Kafka record key, or null if the record has none
    }
}

Listen to topics

# Listen to all topics (one process per topic)
php bin/console rossel:kafka:listen

# Listen to specific topics (comma-separated topic values)
php bin/console rossel:kafka:listen --topics=public_subscription_input_v1_json_delete,public_offer_input_v1_json_delete

Without --topics, configured topics that cannot be consumed are skipped with a warning. See Topic roles.

A message that fails (consumer exception, invalid headers) is logged as an error and skipped: the listener keeps running.

After each message, one EXEC_SUCCESS or EXEC_ERROR log message is sent to the topic, unless producer.send_log_messages is false.

Offsets are committed once a message is processed (at-least-once delivery). If the listener stops while processing a message, that message is read again on restart, so consumers must be idempotent.

Fetch topic metadata

use Rossel\RosselKafka\Enum\Config\Broker\TopicConfigKeys;
use Rossel\RosselKafka\Enum\MessageHeaders\MessageType;

// Get a single topic by config key
$topic = $kafkaTopicsFetcher->get(TopicConfigKeys::KAFKA_TOPIC_PUBLIC_OFFER_OUTPUT_V1_JSON_DELETE);

// Get all topics
$topics = $kafkaTopicsFetcher->getAll();

// Get topics supporting a given message type
$topics = $kafkaTopicsFetcher->getByMessageType(MessageType::SYNC_B2C_ERP_OFFERS);

rossel / rossel-kafka

Package info

Security

Statistics