Add context manager support for Producer, Consumer, and AdminClient

[k-raina]

Summary

This PR adds context manager (with statement) support to Producer, Consumer, and AdminClient classes

Motivation

This addresses most requested community issues Issue #2061 and Issue #1244.

Changes

Implementation

1. Producer Context Manager (src/confluent_kafka/src/Producer.c):

   • Added __enter__ method that returns the producer instance
   • Added __exit__ method that automatically flushes pending messages and destroys the producer
   • Added NULL checks to prevent crashes when methods are called after context exit
   • Flushes all pending messages before destruction to ensure message delivery

2. Consumer Context Manager (src/confluent_kafka/src/Consumer.c):

   • Added __enter__ method that returns the consumer instance
   • Added __exit__ method that calls the existing close() method

3. AdminClient Context Manager (src/confluent_kafka/src/Admin.c):

   • Added __enter__ method that returns the admin client instance
   • Added __exit__ method that destroys the admin client and cleans up resources
   • Added NULL checks in Admin API methods and Metadata operations to prevent use after closure

Type Stubs

• Updated src/confluent_kafka/cimpl.pyi with context manager method signatures for all three classes

Testing

• Added comprehensive unit tests for Producer context manager (tests/test_Producer.py), (tests/test_Consumer.py), (tests/test_Admin.py) covering scenarios
  • Basic usage and return value
  • Exception propagation
  • Behavior after context exit
  • Multiple instances
  • Callback handling during flush

Examples and Documentation

• Updated examples/avro_producer.py to use context managers for both Producer and SchemaRegistryClient
• Added new examples/context_manager_example.py demonstrating all three context managers
• Updated examples/README.md to document the new context manager

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Copilot]

Pull Request Overview

This PR adds context manager support to Producer, Consumer, and AdminClient classes in the confluent-kafka Python library, enabling automatic resource cleanup when exiting with blocks.

Key changes include:

• Context manager implementation for all three client types with automatic cleanup
• Comprehensive test coverage for context manager functionality
• Updated examples demonstrating context manager usage

Reviewed Changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
tests/test_Producer.py	Adds comprehensive unit tests for Producer context manager functionality
tests/test_Consumer.py	Adds unit tests covering Consumer context manager behavior and cleanup
tests/test_Admin.py	Adds unit tests for AdminClient context manager with various Admin APIs
src/confluent_kafka/src/Producer.c	Implements enter and exit methods with automatic flush and NULL checks
src/confluent_kafka/src/Metadata.c	Adds NULL checks to prevent operations on closed handles
src/confluent_kafka/src/Consumer.c	Implements context manager methods that call existing close() functionality
src/confluent_kafka/src/Admin.c	Implements context manager methods with automatic cleanup and NULL checks
src/confluent_kafka/cimpl.pyi	Updates type stubs with context manager method signatures
examples/context_manager_example.py	New comprehensive example demonstrating all three context managers
examples/avro_producer.py	Updates existing example to use context managers
examples/README.md	Documents the new context manager example

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Using an infinite timeout (-1) for flush during context exit could cause the application to hang indefinitely if there are network issues. Consider using a reasonable timeout with a warning message if flush times out.

[Copilot]

The avro_serializer is created inside the SchemaRegistryClient context but used in the Producer context. This creates a dependency where the serializer references a potentially closed schema registry client. Consider restructuring to ensure the schema registry client remains open for the lifetime of the serializer.

[sonarqube-confluent]

Quality Gate failed

Failed conditions
4 New issues
0.0% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

Catch issues before they fail your Quality Gate with our IDE extension SonarQube for IDE