Laravel Kafka Laravel Package

title: SASL Authentication weight: 3

<x-sponsors.request-sponsor/>

SASL allows your producers and your consumers to authenticate to your Kafka cluster, which verifies their identity. It's also a secure way to enable your clients to endorse an identity. To provide SASL configuration, you can use the withSasl method, passing a Junges\Kafka\Config\Sasl instance as the argument:

$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withSasl(
        password: 'password',
        username: 'username',
        mechanisms: 'authentication mechanism'
    );

You can also set the security protocol used with sasl. It's optional and by default SASL_PLAINTEXT is used, but you can set it to SASL_SSL:

$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withSasl(
        password: 'password',
        username: 'username',
        mechanisms: 'authentication mechanism',
        securityProtocol: 'SASL_SSL',
    );

<x-docs.tip title="Hot tip!">
    When using the `withSasl` method, the securityProtocol set in this method takes priority over `withSecurityProtocol` method.
</x-docs.tip>

OAUTHBEARER Authentication

If your Kafka cluster requires OAuth 2.0 / OAUTHBEARER authentication (common with Confluent Cloud, AWS MSK with IAM, or enterprise deployments), you can use the withOAuthBearerTokenRefreshCallback method. This registers a callback that librdkafka invokes whenever it needs a fresh token.

use Junges\Kafka\Facades\Kafka;

$consumer = Kafka::consumer(['my.topic'])
    ->withOptions([
        'security.protocol' => 'SASL_SSL',
        'sasl.mechanisms'   => 'OAUTHBEARER',
    ])
    ->withOAuthBearerTokenRefreshCallback(function ($consumer, string $oauthConfig): void {
        $token      = fetchTokenFromIdP();
        $expiresMs  = getTokenExpiryMs($token);
        $principal   = 'my-client-id';
        $extensions = [
            'logicalCluster' => 'lkc-xxxxx',
            'identityPoolId' => 'pool-xxxxx',
        ];

        $consumer->oauthbearerSetToken($token, $expiresMs, $principal, $extensions);
    })
    ->withHandler(new MyMessageHandler())
    ->build()
    ->consume();

The callback receives two arguments: the RdKafka\KafkaConsumer (or RdKafka\Producer) instance and the oauthbearer_config string from your librdkafka configuration. Inside the callback, call $consumer->oauthbearerSetToken() to provide the token, or $consumer->oauthbearerSetTokenFailure($reason) if the token could not be obtained.

This method is available on both the consumer and producer builders.

TLS Authentication

For using TLS authentication with Laravel Kafka you can configure your client using the following options:

$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withOptions([
        'ssl.ca.location' => '/some/location/kafka.crt',
        'ssl.certificate.location' => '/some/location/client.crt',
        'ssl.key.location' => '/some/location/client.key',
        'ssl.endpoint.identification.algorithm' => 'none'
    ]);