Overview

confluent-kafka 1.7, released on May 16, 2021, adds OAuth authentication support and improves error handling.

Main Features

OAuth support

The client now supports OAuth/OIDC authentication via a token refresh callback, enabling secure integration with identity providers.

python 
from confluent_kafka import Producer

def oauth_callback(config):
    """Callback to obtain an OAuth token."""
    # Call to the identity provider
    return 'jwt_token_here', expiry_time

producer = Producer({
    'bootstrap.servers': 'kafka:9092',
    'security.protocol': 'SASL_SSL',
    'sasl.mechanism': 'OAUTHBEARER',
    'oauth_cb': oauth_callback,
})

producer.produce('my-topic', value=b'message')
producer.flush()

Improved error handling

Error handling is strengthened with more detailed error callbacks and better distinction between fatal and recoverable errors.

python 
from confluent_kafka import Consumer, KafkaError

def error_callback(err):
    if err.code() == KafkaError._ALL_BROKERS_DOWN:
        print('All brokers are down')
    elif err.fatal():
        raise SystemExit(f'Fatal error: {err}')
    else:
        print(f'Recoverable error: {err}')

consumer = Consumer({
    'bootstrap.servers': 'kafka:9092',
    'group.id': 'my-group',
    'error_cb': error_callback,
})

Sources