<artifacts>
<artifact type="text/markdown" title="Updated dagster-kafka README.md with Confluent Connect Integration">
## Dagster Kafka Integration
[](https://badge.fury.io/py/dagster-kafka)
[](https://pypi.org/project/dagster-kafka/)
[](https://pepy.tech/project/dagster-kafka)
[](https://opensource.org/licenses/Apache-2.0)
**The most comprehensively validated Kafka integration for Dagster** - Supporting all four major serialization formats with enterprise-grade features, complete security, operational tooling, and YAML-based Components.
## What's New in v1.3.1
### Confluent Connect Integration (NEW)
- **Complete Connect Management**: Native REST API integration for Kafka Connect
- **Connector Assets**: Define Kafka connectors as Dagster Software-Defined Assets
- **Automated Health Monitoring**: Intelligent connector monitoring with auto-recovery
- **Enterprise Operations**: CLI tools for connector management and monitoring
- **Production Ready**: Thoroughly tested with race condition handling and load testing
### JSON Schema Validation
- **4th Serialization Format**: Complete JSON Schema validation support
- **Data Quality Enforcement**: Automatic validation with configurable strictness
- **Schema Evolution**: Track and validate schema changes over time
- **Enterprise DLQ Integration**: Invalid data automatically routed to Dead Letter Queue
- **Production Ready**: Circuit breaker patterns and comprehensive error handling
## 📋 Table of Contents
- [Installation](#installation)
- [Quick Start](#quick-start)
- [Serialization Formats](#serialization-formats)
- [JSON Schema Validation](#json-schema-validation)
- [JSON Support](#json-support)
- [Avro Support](#avro-support)
- [Protobuf Support](#protobuf-support)
- [Confluent Connect Integration](#confluent-connect-integration)
- [Connect Management](#connect-management)
- [Connector Assets](#connector-assets)
- [Health Monitoring](#health-monitoring)
- [Recovery Patterns](#recovery-patterns)
- [CLI Tools](#cli-tools)
- [Enterprise Features](#enterprise-features)
- [Dead Letter Queue (DLQ)](#dead-letter-queue-dlq)
- [Security](#security)
- [Performance](#performance)
- [Examples](#examples)
- [Development](#development)
- [Contributing](#contributing)
## Installation
```bash
pip install dagster-kafka
```
**Requirements**: Python 3.9+ | Dagster 1.5.0+
## Quick Start
### Confluent Connect Integration (New)
```python
from dagster import Definitions
from dagster_kafka.connect import ConfluentConnectResource, create_connector_asset
# Define a connector as a Dagster asset
mirror_connector = create_connector_asset(
group_name="kafka_connect",
)
# Define your Dagster job with the connector asset
defs = Definitions(
assets=[mirror_connector],
resources={
"connect": ConfluentConnectResource(
connect_url="http://localhost:8083",
)
},
)
# Use with configuration
"""
ops:
connector_asset:
config:
name: "my-source-connector"
connector_class: "org.apache.kafka.connect.file.FileStreamSourceConnector"
config:
tasks.max: "1"
file: "/tmp/test-source.txt"
topic: "test-topic"
"""
```
### JSON Schema Validation (Recommended)
```python
from dagster import asset, Definitions
from dagster_kafka import KafkaResource, create_json_schema_kafka_io_manager, DLQStrategy
# Define your data quality schema
user_events_schema = {
"type": "object",
"properties": {
"user_id": {"type": "string"},
"event_type": {"type": "string", "enum": ["login", "logout", "click"]},
"timestamp": {"type": "string", "format": "date-time"},
"metadata": {
"type": "object",
"properties": {
"ip_address": {"type": "string"},
"user_agent": {"type": "string"}
},
"required": ["ip_address"]
}
},
"required": ["user_id", "event_type", "timestamp"]
}
@asset(io_manager_key="json_schema_io_manager")
def validated_user_events():
"""Consume user events with automatic JSON Schema validation."""
pass
defs = Definitions(
assets=[validated_user_events],
resources={
"kafka": KafkaResource(bootstrap_servers="localhost:9092"),
"json_schema_io_manager": create_json_schema_kafka_io_manager(
kafka_resource=KafkaResource(bootstrap_servers="localhost:9092"),
schema_dict=user_events_schema,
enable_schema_validation=True,
strict_validation=True,
enable_dlq=True,
dlq_strategy=DLQStrategy.CIRCUIT_BREAKER
)
}
)
```
## Serialization Formats
This integration supports **all four major serialization formats** used in modern data engineering:
| Format | Schema Support | Validation | Registry | Performance | Best For |
|--------|---------------|------------|----------|-------------|----------|
| **JSON** | ❌ | ❌ | ❌ | Good | Simple events, logs |
| **JSON Schema** | ✅ | ✅ | ❌ | Good | **Data quality enforcement** |
| **Avro** | ✅ | ✅ | ✅ | Better | Schema evolution, analytics |
| **Protobuf** | ✅ | ✅ | ✅ | Best | High-performance, microservices |
### JSON Schema Validation
**Enforce data quality with automatic validation**
#### Features
- **Automatic Validation**: Messages validated against JSON Schema on consumption
- **Flexible Modes**: Strict (fail on invalid) or lenient (warn and continue)
- **Schema Evolution**: Track schema changes and compatibility
- **DLQ Integration**: Invalid messages automatically routed for investigation
- **File or Inline**: Load schemas from files or define inline
#### Basic Usage
```python
from dagster_kafka import create_json_schema_kafka_io_manager
# Using schema file
json_schema_manager = create_json_schema_kafka_io_manager(
kafka_resource=kafka_resource,
schema_file="schemas/user_events.json",
enable_schema_validation=True,
strict_validation=True
)
# Using inline schema
json_schema_manager = create_json_schema_kafka_io_manager(
kafka_resource=kafka_resource,
schema_dict={
"type": "object",
"properties": {
"id": {"type": "string"},
"timestamp": {"type": "string", "format": "date-time"}
},
"required": ["id", "timestamp"]
},
enable_schema_validation=True
)
```
### JSON Support
Basic JSON message consumption without schema validation.
```python
from dagster_kafka import KafkaIOManager
json_manager = KafkaIOManager(
kafka_resource=kafka_resource,
consumer_group_id="json-consumer",
enable_dlq=True
)
```
### Avro Support
Binary format with Schema Registry integration and evolution validation.
```python
from dagster_kafka import avro_kafka_io_manager
@asset(io_manager_key="avro_kafka_io_manager")
def user_data(context, config):
"""Load user events using Avro schema with validation."""
io_manager = context.resources.avro_kafka_io_manager
return io_manager.load_input(
context,
topic="user-events",
schema_file="schemas/user.avsc",
validate_evolution=True
)
```
### Protobuf Support
High-performance binary format with full schema management.
```python
from dagster_kafka.protobuf_io_manager import create_protobuf_kafka_io_manager
protobuf_manager = create_protobuf_kafka_io_manager(
kafka_resource=kafka_resource,
schema_registry_url="http://localhost:8081",
consumer_group_id="protobuf-consumer",
enable_dlq=True
)
```
## Confluent Connect Integration
The Confluent Connect integration provides a complete solution for managing Kafka Connect connectors within your Dagster environment.
### Connect Management
**REST API Client for Kafka Connect**
```python
from dagster_kafka.connect import ConfluentConnectClient
# Create a client to interact with the Connect REST API
client = ConfluentConnectClient(base_url="http://localhost:8083")
# List available connectors
connectors = client.list_connectors()
# Get connector status
status = client.get_connector_status("my-connector")
# Create a new connector
connector_config = {
"name": "file-source-connector",
"config": {
"connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
"tasks.max": "1",
"file": "/tmp/test-source.txt",
"topic": "test-topic"
}
}
client.create_connector(connector_config)
# Update configuration
client.update_connector_config("file-source-connector", {"tasks.max": "2"})
# Manage connector lifecycle
client.pause_connector("file-source-connector")
client.resume_connector("file-source-connector")
client.restart_connector("file-source-connector")
```
### Connector Assets
**Define Kafka Connect connectors as Dagster assets**
```python
from dagster import Definitions
from dagster_kafka.connect import ConfluentConnectResource, create_connector_asset
# Create a connector asset
source_connector = create_connector_asset(
group_name="kafka_connect",
)
# Define your Dagster job
defs = Definitions(
assets=[source_connector],
resources={
"connect": ConfluentConnectResource(
connect_url="http://localhost:8083",
)
},
)
```
#### Configuration Example
```yaml
ops:
connector_asset:
config:
name: "mysql-source-connector"
connector_class: "io.debezium.connector.mysql.MySqlConnector"
config:
tasks.max: "1"
database.hostname: "mysql"
database.port: "3306"
database.user: "debezium"
database.password: "dbz"
database.server.id: "184054"
database.server.name: "dbserver1"
database.include.list: "inventory"
database.history.kafka.bootstrap.servers: "kafka:9092"
database.history.kafka.topic: "schema-changes.inventory"
```
### Health Monitoring
**Monitor connector health and implement auto-recovery**
```python
from dagster import job, op, sensor, SensorResult, RunRequest
from dagster_kafka.connect import ConfluentConnectResource
@sensor(
name="connect_health_sensor",
minimum_interval_seconds=60,
required_resource_keys={"connect"},
)
def connector_health_sensor(context):
"""Monitor the health of Kafka Connect connectors."""
connect = context.resources.connect
connector_names = ["mysql-source", "elasticsearch-sink"]
unhealthy_connectors = []
# Check each connector
for connector_name in connector_names:
try:
status = connect.get_connector_status(connector_name)
connector_state = status["connector"]["state"]
# Check if connector is unhealthy
if connector_state != "RUNNING":
context.log.warning(
f"Connector {connector_name} is in {connector_state} state"
)
unhealthy_connectors.append({
"name": connector_name,
"state": connector_state
})
# Also check tasks
for task in status.get("tasks", []):
task_state = task.get("state")
task_id = task.get("id")
if task_state != "RUNNING":
context.log.warning(
f"Task {task_id} of connector {connector_name} is in {task_state} state"
)
unhealthy_connectors.append({
"name": connector_name,
"task_id": task_id,
"state": task_state
})
except Exception as e:
context.log.error(f"Error checking connector {connector_name}: {e}")
# If we have unhealthy connectors, trigger a remediation job
if unhealthy_connectors:
return RunRequest(
run_key=f"connector_health_{context.cursor}",
job_name="remediate_connectors_job",
run_config={
"ops": {
"remediate_connectors": {
"config": {
"unhealthy_connectors": unhealthy_connectors
}
}
}
},
)
return SensorResult(
skip_reason="All connectors healthy",
cursor=str(context.get_current_time()),
)
```
### Recovery Patterns
**Implement automated recovery from connector failures**
```python
from dagster_kafka.connect import ConfluentConnectClient
def advanced_recovery(client, connector_name, max_attempts=3):
"""
Advanced recovery pattern that tries different recovery methods
based on the nature of the failure.
"""
for attempt in range(1, max_attempts + 1):
try:
# Check current status
status = client.get_connector_status(connector_name)
connector_state = status["connector"]["state"]
if connector_state == "RUNNING":
# Check if any tasks are failing
tasks = status.get("tasks", [])
failing_tasks = [
task for task in tasks
if task.get("state") != "RUNNING"
]
if not failing_tasks:
return True
# Restart only the failing tasks
for task in failing_tasks:
task_id = task.get("id")
client.restart_task(connector_name, int(task_id))
else:
# Try different recovery strategies based on state
if connector_state == "FAILED":
client.restart_connector(connector_name)
elif connector_state == "PAUSED":
client.resume_connector(connector_name)
else:
client.restart_connector(connector_name)
client.resume_connector(connector_name)
# Wait for recovery to take effect and check again
time.sleep(5)
status = client.get_connector_status(connector_name)
if status["connector"]["state"] == "RUNNING":
return True
except Exception as e:
pass
return False
```
### CLI Tools
The Confluent Connect integration includes several command-line tools for connector management and monitoring:
```bash
# List all connectors
python connect_cli.py list
# Get connector status
python connect_cli.py status mysql-connector
# Create a new connector
python connect_cli.py create connector_config.json
# Update connector configuration
python connect_cli.py update mysql-connector updated_config.json
# Restart a connector
python connect_cli.py restart mysql-connector
# Monitor connector health
python connect_health_monitor.py --auto-restart mysql-connector elasticsearch-connector
```
## Enterprise Features
### Comprehensive Enterprise Validation
**Version 1.3.0** - Most validated Kafka integration package ever created:
#### 12-Phase Enterprise Validation Completed
- **EXCEPTIONAL Performance**: 1,199 messages/second peak throughput
- **Security Hardened**: Complete credential validation + network security
- **Stress Tested**: 100% success rate (305/305 operations over 8+ minutes)
- **Memory Efficient**: Stable under extended load (+42MB over 8 minutes)
- **Enterprise Ready**: Complete DLQ tooling suite with 5 CLI tools
- **Zero Critical Issues**: Across all validation phases
- **Connect Integration Validated**: Race condition and load testing complete
- **JSON Schema Validated**: 4th serialization format thoroughly tested
#### Validation Results Summary
| Phase | Test Type | Result | Key Metrics |
|-------|-----------|--------|-------------|
| **Phase 5** | Performance Testing | ✅ **PASS** | 1,199 msgs/sec peak throughput |
| **Phase 7** | Integration Testing | ✅ **PASS** | End-to-end message flow validated |
| **Phase 9** | Compatibility Testing | ✅ **PASS** | Python 3.12 + Dagster 1.11.3 |
| **Phase 10** | Security Audit | ✅ **PASS** | Credential + network security |
| **Phase 11** | Stress Testing | ✅ **EXCEPTIONAL** | 100% success rate, 305 operations |
| **Phase 12** | Connect Integration | ✅ **PASS** | Race condition and load testing |
### Core Enterprise Features
- **Complete Security**: SASL/SSL authentication and encryption
- **Schema Evolution**: Breaking change detection across all formats
- **Production Monitoring**: Real-time alerting with Slack/Email integration
- **High Performance**: Advanced caching, batching, and connection pooling
- **Error Recovery**: Multiple recovery strategies for production resilience
- **Dagster Components**: YAML-based configuration for teams
- **Connect Integration**: Complete Kafka Connect management
### Enterprise DLQ Tooling Suite
Complete operational tooling for Dead Letter Queue management:
```bash
# Analyze failed messages with comprehensive error pattern analysis
dlq-inspector --topic user-events --max-messages 20
# Replay messages with filtering and safety controls
dlq-replayer --source-topic orders_dlq --target-topic orders --dry-run
# Monitor DLQ health across multiple topics
dlq-monitor --topics user-events_dlq,orders_dlq --output-format json
# Set up automated alerting
dlq-alerts --topic critical-events_dlq --max-messages 500
# Operations dashboard for DLQ health monitoring
dlq-dashboard --topics user-events_dlq,orders_dlq
```
## Dead Letter Queue (DLQ)
### DLQ Strategies
- **DISABLED**: No DLQ processing
- **IMMEDIATE**: Send to DLQ immediately on failure
- **RETRY_THEN_DLQ**: Retry N times, then send to DLQ
- **CIRCUIT_BREAKER**: Circuit breaker pattern with DLQ fallback
### Error Classification
- **DESERIALIZATION_ERROR**: Failed to deserialize message
- **SCHEMA_ERROR**: Schema validation failed (includes JSON Schema validation)
- **PROCESSING_ERROR**: Business logic error
- **CONNECTION_ERROR**: Kafka connection issues
- **TIMEOUT_ERROR**: Message processing timeout
- **UNKNOWN_ERROR**: Unclassified errors
### Circuit Breaker Configuration
```python
from dagster_kafka import DLQConfiguration, DLQStrategy
dlq_config = DLQConfiguration(
strategy=DLQStrategy.CIRCUIT_BREAKER,
circuit_breaker_failure_threshold=5, # Open after 5 failures
circuit_breaker_recovery_timeout_ms=30000, # Test recovery after 30s
circuit_breaker_success_threshold=2 # Close after 2 successes
)
```
## Security
### Security Protocols Supported
- **PLAINTEXT**: For local development and testing
- **SSL**: Certificate-based encryption
- **SASL_PLAINTEXT**: Username/password authentication
- **SASL_SSL**: Combined authentication and encryption (recommended for production)
### SASL Authentication Mechanisms
- **PLAIN**: Simple username/password authentication
- **SCRAM-SHA-256**: Secure challenge-response authentication
- **SCRAM-SHA-512**: Enhanced secure authentication
- **GSSAPI**: Kerberos authentication for enterprise environments
- **OAUTHBEARER**: OAuth-based authentication
### Secure Production Example
```python
from dagster_kafka import KafkaResource, SecurityProtocol, SaslMechanism
secure_kafka = KafkaResource(
bootstrap_servers="prod-kafka-01:9092,prod-kafka-02:9092",
security_protocol=SecurityProtocol.SASL_SSL,
sasl_mechanism=SaslMechanism.SCRAM_SHA_256,
sasl_username="production-user",
sasl_password="secure-password",
ssl_ca_location="/etc/ssl/certs/kafka-ca.pem",
ssl_check_hostname=True
)
```
## Performance
### Validated Performance Results
- **Peak Throughput**: 1,199 messages/second
- **Stress Test Success**: 100% (305/305 operations)
- **Extended Stability**: 8+ minutes continuous operation
- **Memory Efficiency**: +42MB over extended load (excellent)
- **Concurrent Operations**: 120/120 successful operations
- **Resource Management**: Zero thread accumulation
### Enterprise Stability Testing
```
PASS Extended Stability: 5+ minutes, 137/137 successful materializations
PASS Resource Management: 15 cycles, no memory leaks detected
PASS Concurrent Usage: 8 threads × 15 operations = 100% success
PASS Comprehensive Stress: 8+ minutes, 305 operations, EXCEPTIONAL rating
```
## Examples
### Complete Kafka Ecosystem Integration
```python
from dagster import Definitions, asset, AssetExecutionContext
from dagster_kafka import KafkaResource
from dagster_kafka.connect import ConfluentConnectResource, create_connector_asset
from dagster_kafka.json_schema_io_manager import create_json_schema_kafka_io_manager
# Source connector that ingests data into Kafka
source_connector = create_connector_asset(
key_prefix=["mysql", "cdc"],
group_name="source_connectors",
)
# Processing asset that consumes and transforms data
@asset(
key_prefix=["kafka", "processed"],
group_name="processing",
deps=[source_connector], # Depends on source connector
io_manager_key="json_schema_io_manager",
)
def transform_data(context: AssetExecutionContext):
"""Transform the data from the source topic."""
# Your transformation logic
return {"transformed": "data"}
# Sink connector that exports data to external systems
sink_connector = create_connector_asset(
key_prefix=["elasticsearch", "sink"],
group_name="sink_connectors",
deps=[transform_data], # Depends on the transformation
)
# Define your complete pipeline with Kafka ecosystem integration
defs = Definitions(
assets=[source_connector, transform_data, sink_connector],
resources={
# Kafka resource for message consumption/production
"kafka": KafkaResource(
bootstrap_servers="localhost:9092",
),
# Connect resource for connector management
"connect": ConfluentConnectResource(
connect_url="http://localhost:8083",
),
# JSON Schema validation for data quality
"json_schema_io_manager": create_json_schema_kafka_io_manager(
kafka_resource=KafkaResource(bootstrap_servers="localhost:9092"),
schema_file="schemas/event_schema.json",
enable_schema_validation=True,
strict_validation=True,
),
},
)
```
### Connector Health Monitoring Integration
```python
from dagster import Definitions, job, op, sensor, Config
from dagster_kafka.connect import ConfluentConnectResource
class UnhealthyConnectorsConfig(Config):
unhealthy_connectors: List[Dict[str, Any]]
@op
def remediate_connectors(context, config: UnhealthyConnectorsConfig):
"""Auto-remediate unhealthy connectors."""
connect = context.resources.connect
unhealthy_connectors = config.unhealthy_connectors
for connector in unhealthy_connectors:
connector_name = connector["name"]
# If it's a task issue, restart the specific task
if "task_id" in connector:
task_id = connector["task_id"]
context.log.info(f"Restarting task {task_id} of connector {connector_name}")
connect.restart_task(connector_name, int(task_id))
# Otherwise restart the entire connector
else:
context.log.info(f"Restarting connector {connector_name}")
connect.restart_connector(connector_name)
# Resume if it was paused
context.log.info(f"Resuming connector {connector_name}")
connect.resume_connector(connector_name)
return {
"remediated_count": len(unhealthy_connectors),
"connector_names": [c["name"] for c in unhealthy_connectors]
}
@job
def remediate_connectors_job():
"""Job to remediate unhealthy connectors."""
remediate_connectors()
# Define your Dagster components
defs = Definitions(
jobs=[remediate_connectors_job],
sensors=[connector_health_sensor], # Use the sensor defined earlier
resources={
"connect": ConfluentConnectResource(
connect_url="http://localhost:8083",
)
},
)
```
## Development
### Running Tests
```bash
# Run all validation tests (12 phases)
python -m pytest tests/ -v
# Specific test modules
python -m pytest tests/test_connect_client.py -v # Connect client tests
python -m pytest tests/test_connect_resource.py -v # Connect resource tests
python -m pytest tests/test_connect_assets.py -v # Connect assets tests
python -m pytest tests/test_json_schema_io_manager.py -v # JSON Schema tests
python -m pytest tests/test_avro_io_manager.py -v # Avro tests
python -m pytest tests/test_protobuf_io_manager.py -v # Protobuf tests
python -m pytest tests/test_dlq.py -v # DLQ tests
python -m pytest tests/test_security.py -v # Security tests
python -m pytest tests/test_performance.py -v # Performance tests
```
### Local Development Setup
```bash
# Clone the repository
git clone https://github.com/kingsley-123/dagster-kafka-integration.git
cd dagster-kafka-integration
# Install dependencies
pip install -r requirements.txt
# Install in development mode
pip install -e .
# Start local Kafka and Connect for testing
docker-compose up -d
```
### Example Directory Structure
```
examples/
├── json_examples/ # Basic JSON message examples
├── json_schema_examples/ # JSON Schema validation examples
├── avro_examples/ # Avro schema examples
├── protobuf_examples/ # Protobuf examples
├── connect_examples/ # Confluent Connect integration examples (NEW)
├── components_examples/ # YAML Components configuration
├── dlq_examples/ # Complete DLQ tooling suite
├── security_examples/ # Enterprise security examples
├── performance_examples/ # Performance optimization
└── production_examples/ # Enterprise deployment patterns
```
## Why Choose This Integration
### Complete Solution
- **Only integration supporting all 4 major formats** (JSON, JSON Schema, Avro, Protobuf)
- **Complete Kafka ecosystem integration** with Confluent Connect support
- **Enterprise-grade security** with SASL/SSL support
- **Production-ready** with comprehensive monitoring
- **Advanced error handling** with Dead Letter Queue support
- **Complete DLQ Tooling Suite** for enterprise operations
### Developer Experience
- **Multiple configuration options** - Python API OR simple YAML Components
- **Team accessibility** - Components enable non-Python users to configure assets
- **Familiar Dagster patterns** - feels native to the platform
- **Comprehensive examples** for all use cases including security and DLQ
- **Extensive documentation** and testing
- **Production-ready CLI tooling** for DLQ management and Connect operations
### Enterprise Ready
- **12-phase comprehensive validation** covering all scenarios
- **Real-world deployment patterns** and examples
- **Performance optimization** tools and monitoring
- **Enterprise security** for production Kafka clusters
- **Bulletproof error handling** with circuit breaker patterns
- **Complete operational tooling** for DLQ management and Connect integration
### Unprecedented Validation
- **Most validated package** in the Dagster ecosystem
- **Performance proven**: 1,199 msgs/sec peak throughput
- **Stability proven**: 100% success rate under stress
- **Security proven**: Complete credential and network validation
- **Enterprise proven**: Exceptional rating across all dimensions
## Roadmap
### Completed Features (v1.3.0)
- **JSON Support** - Complete native integration ✅
- **JSON Schema Support** - Data validation with evolution checking ✅
- **Avro Support** - Full Schema Registry + evolution validation ✅
- **Protobuf Support** - Complete Protocol Buffers integration ✅
- **Confluent Connect Integration** - Complete Connect REST API integration ✅
- **Connector Assets** - Define connectors as Dagster assets ✅
- **Health Monitoring** - Automated connector health monitoring ✅
- **Recovery Patterns** - Advanced connector recovery strategies ✅
- **Connect CLI Tools** - Complete command-line tools for Connect operations ✅
- **Dagster Components** - YAML-based configuration support ✅
- **Enterprise Security** - Complete SASL/SSL authentication and encryption ✅
- **Schema Evolution** - All compatibility levels across formats ✅
- **Production Monitoring** - Real-time alerting and metrics ✅
- **High-Performance Optimization** - Caching, batching, pooling ✅
- **Dead Letter Queues** - Advanced error handling with circuit breaker ✅
- **Complete DLQ Tooling Suite** - Inspector, Replayer, Monitoring, Alerting ✅
- **Comprehensive Testing** - 12-phase enterprise validation ✅
- **PyPI Distribution** - Official package published and validated ✅
- **Security Hardening** - Configuration injection protection ✅
### Upcoming Features
- **Enhanced JSON Schema** - Schema registry integration
- **Advanced Connect Monitoring** - Custom metrics and dashboards
- **Connect Templates** - Pre-built connector configurations
## Contributing
Contributions are welcome! This project aims to be the definitive Kafka integration for Dagster.
### Ways to contribute:
- **Report issues** - Found a bug? Let us know!
- **Feature requests** - What would make this more useful?
- **Documentation** - Help improve examples and guides
- **Code contributions** - PRs welcome for any improvements
- **Security testing** - Help test security configurations
- **DLQ testing** - Help test error handling scenarios
- **Connect testing** - Help test connector integration scenarios
## License
Apache 2.0 - see [LICENSE](LICENSE) file for details.
## Community & Support
- **GitHub Issues**: [Report bugs and request features](https://github.com/kingsley-123/dagster-kafka-integration/issues)
- **GitHub Discussions**: [Share use cases and get help](https://github.com/kingsley-123/dagster-kafka-integration/discussions)
- **PyPI Package**: [Install and documentation](https://pypi.org/project/dagster-kafka/)
- **Star the repo**: If this helped your project!
## Acknowledgments
- **Dagster Community**: For the initial feature request and continued feedback
- **Contributors**: Thanks to all who provided feedback, testing, and code contributions
- **Enterprise Users**: Built in response to real production deployment needs
- **Security Community**: Special thanks for security testing and validation
- **JSON Schema Community**: Thanks for validation methodology and best practices
- **Confluent Community**: For guidance on Connect integration best practices
---
## The Complete Enterprise Solution
**The most comprehensively validated Kafka integration for Dagster** - supporting all four major serialization formats (JSON, JSON Schema, Avro, Protobuf) with enterprise-grade production features, complete security, advanced Dead Letter Queue error handling, Confluent Connect integration, YAML-based Components, and complete operational tooling suite.
**Version 1.3.1** - Confluent Connect Integration Release
*Built by [Kingsley Okonkwo](https://github.com/kingsley-123) - Solving real data engineering problems with comprehensive open source solutions.*
</artifact>
</artifacts>
Raw data
{
"_id": null,
"home_page": "https://github.com/kingsley-123/dagster-kafka-integration",
"name": "dagster-kafka",
"maintainer": "Kingsley Okonkwo",
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": "kingskonk@gmail.com",
"keywords": "dagster, kafka, apache-kafka, streaming, data-engineering, data-pipeline, etl, data-processing, json, json-schema, avro, protobuf, serialization, enterprise, production, monitoring, alerting, dlq, dead-letter-queue, error-handling, circuit-breaker, microservices, distributed-systems, real-time, schema-registry, confluent, data-validation, sasl, ssl, security, authentication, authorization, confluent-connect, kafka-connect, connectors, cdc",
"author": "Kingsley Okonkwo",
"author_email": "kingskonk@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/6a/82/27caef0264da7ae7246c91142c647ad028e7e5034c1b9c6000bbbb3432a0/dagster_kafka-1.3.1.tar.gz",
"platform": "any",
"description": "<artifacts>\r\n<artifact type=\"text/markdown\" title=\"Updated dagster-kafka README.md with Confluent Connect Integration\">\r\n\r\n## Dagster Kafka Integration\r\n\r\n[](https://badge.fury.io/py/dagster-kafka)\r\n[](https://pypi.org/project/dagster-kafka/)\r\n[](https://pepy.tech/project/dagster-kafka)\r\n[](https://opensource.org/licenses/Apache-2.0)\r\n\r\n**The most comprehensively validated Kafka integration for Dagster** - Supporting all four major serialization formats with enterprise-grade features, complete security, operational tooling, and YAML-based Components.\r\n\r\n## What's New in v1.3.1\r\n\r\n### Confluent Connect Integration (NEW)\r\n- **Complete Connect Management**: Native REST API integration for Kafka Connect\r\n- **Connector Assets**: Define Kafka connectors as Dagster Software-Defined Assets\r\n- **Automated Health Monitoring**: Intelligent connector monitoring with auto-recovery\r\n- **Enterprise Operations**: CLI tools for connector management and monitoring\r\n- **Production Ready**: Thoroughly tested with race condition handling and load testing\r\n\r\n### JSON Schema Validation\r\n- **4th Serialization Format**: Complete JSON Schema validation support\r\n- **Data Quality Enforcement**: Automatic validation with configurable strictness\r\n- **Schema Evolution**: Track and validate schema changes over time\r\n- **Enterprise DLQ Integration**: Invalid data automatically routed to Dead Letter Queue\r\n- **Production Ready**: Circuit breaker patterns and comprehensive error handling\r\n\r\n## \ud83d\udccb Table of Contents\r\n\r\n- [Installation](#installation)\r\n- [Quick Start](#quick-start)\r\n- [Serialization Formats](#serialization-formats)\r\n - [JSON Schema Validation](#json-schema-validation)\r\n - [JSON Support](#json-support)\r\n - [Avro Support](#avro-support)\r\n - [Protobuf Support](#protobuf-support)\r\n- [Confluent Connect Integration](#confluent-connect-integration)\r\n - [Connect Management](#connect-management)\r\n - [Connector Assets](#connector-assets)\r\n - [Health Monitoring](#health-monitoring)\r\n - [Recovery Patterns](#recovery-patterns)\r\n - [CLI Tools](#cli-tools)\r\n- [Enterprise Features](#enterprise-features)\r\n- [Dead Letter Queue (DLQ)](#dead-letter-queue-dlq)\r\n- [Security](#security)\r\n- [Performance](#performance)\r\n- [Examples](#examples)\r\n- [Development](#development)\r\n- [Contributing](#contributing)\r\n\r\n## Installation\r\n\r\n```bash\r\npip install dagster-kafka\r\n```\r\n\r\n**Requirements**: Python 3.9+ | Dagster 1.5.0+\r\n\r\n## Quick Start\r\n\r\n### Confluent Connect Integration (New)\r\n\r\n```python\r\nfrom dagster import Definitions\r\nfrom dagster_kafka.connect import ConfluentConnectResource, create_connector_asset\r\n\r\n# Define a connector as a Dagster asset\r\nmirror_connector = create_connector_asset(\r\n group_name=\"kafka_connect\",\r\n)\r\n\r\n# Define your Dagster job with the connector asset\r\ndefs = Definitions(\r\n assets=[mirror_connector],\r\n resources={\r\n \"connect\": ConfluentConnectResource(\r\n connect_url=\"http://localhost:8083\",\r\n )\r\n },\r\n)\r\n\r\n# Use with configuration\r\n\"\"\"\r\nops:\r\n connector_asset:\r\n config:\r\n name: \"my-source-connector\"\r\n connector_class: \"org.apache.kafka.connect.file.FileStreamSourceConnector\"\r\n config:\r\n tasks.max: \"1\"\r\n file: \"/tmp/test-source.txt\"\r\n topic: \"test-topic\"\r\n\"\"\"\r\n```\r\n\r\n### JSON Schema Validation (Recommended)\r\n```python\r\nfrom dagster import asset, Definitions\r\nfrom dagster_kafka import KafkaResource, create_json_schema_kafka_io_manager, DLQStrategy\r\n\r\n# Define your data quality schema\r\nuser_events_schema = {\r\n \"type\": \"object\",\r\n \"properties\": {\r\n \"user_id\": {\"type\": \"string\"},\r\n \"event_type\": {\"type\": \"string\", \"enum\": [\"login\", \"logout\", \"click\"]},\r\n \"timestamp\": {\"type\": \"string\", \"format\": \"date-time\"},\r\n \"metadata\": {\r\n \"type\": \"object\",\r\n \"properties\": {\r\n \"ip_address\": {\"type\": \"string\"},\r\n \"user_agent\": {\"type\": \"string\"}\r\n },\r\n \"required\": [\"ip_address\"]\r\n }\r\n },\r\n \"required\": [\"user_id\", \"event_type\", \"timestamp\"]\r\n}\r\n\r\n@asset(io_manager_key=\"json_schema_io_manager\")\r\ndef validated_user_events():\r\n \"\"\"Consume user events with automatic JSON Schema validation.\"\"\"\r\n pass\r\n\r\ndefs = Definitions(\r\n assets=[validated_user_events],\r\n resources={\r\n \"kafka\": KafkaResource(bootstrap_servers=\"localhost:9092\"),\r\n \"json_schema_io_manager\": create_json_schema_kafka_io_manager(\r\n kafka_resource=KafkaResource(bootstrap_servers=\"localhost:9092\"),\r\n schema_dict=user_events_schema,\r\n enable_schema_validation=True,\r\n strict_validation=True,\r\n enable_dlq=True,\r\n dlq_strategy=DLQStrategy.CIRCUIT_BREAKER\r\n )\r\n }\r\n)\r\n```\r\n\r\n## Serialization Formats\r\n\r\nThis integration supports **all four major serialization formats** used in modern data engineering:\r\n\r\n| Format | Schema Support | Validation | Registry | Performance | Best For |\r\n|--------|---------------|------------|----------|-------------|----------|\r\n| **JSON** | \u274c | \u274c | \u274c | Good | Simple events, logs |\r\n| **JSON Schema** | \u2705 | \u2705 | \u274c | Good | **Data quality enforcement** |\r\n| **Avro** | \u2705 | \u2705 | \u2705 | Better | Schema evolution, analytics |\r\n| **Protobuf** | \u2705 | \u2705 | \u2705 | Best | High-performance, microservices |\r\n\r\n### JSON Schema Validation\r\n\r\n**Enforce data quality with automatic validation**\r\n\r\n#### Features\r\n- **Automatic Validation**: Messages validated against JSON Schema on consumption\r\n- **Flexible Modes**: Strict (fail on invalid) or lenient (warn and continue)\r\n- **Schema Evolution**: Track schema changes and compatibility\r\n- **DLQ Integration**: Invalid messages automatically routed for investigation\r\n- **File or Inline**: Load schemas from files or define inline\r\n\r\n#### Basic Usage\r\n```python\r\nfrom dagster_kafka import create_json_schema_kafka_io_manager\r\n\r\n# Using schema file\r\njson_schema_manager = create_json_schema_kafka_io_manager(\r\n kafka_resource=kafka_resource,\r\n schema_file=\"schemas/user_events.json\",\r\n enable_schema_validation=True,\r\n strict_validation=True\r\n)\r\n\r\n# Using inline schema\r\njson_schema_manager = create_json_schema_kafka_io_manager(\r\n kafka_resource=kafka_resource,\r\n schema_dict={\r\n \"type\": \"object\",\r\n \"properties\": {\r\n \"id\": {\"type\": \"string\"},\r\n \"timestamp\": {\"type\": \"string\", \"format\": \"date-time\"}\r\n },\r\n \"required\": [\"id\", \"timestamp\"]\r\n },\r\n enable_schema_validation=True\r\n)\r\n```\r\n\r\n### JSON Support\r\n\r\nBasic JSON message consumption without schema validation.\r\n\r\n```python\r\nfrom dagster_kafka import KafkaIOManager\r\n\r\njson_manager = KafkaIOManager(\r\n kafka_resource=kafka_resource,\r\n consumer_group_id=\"json-consumer\",\r\n enable_dlq=True\r\n)\r\n```\r\n\r\n### Avro Support\r\n\r\nBinary format with Schema Registry integration and evolution validation.\r\n\r\n```python\r\nfrom dagster_kafka import avro_kafka_io_manager\r\n\r\n@asset(io_manager_key=\"avro_kafka_io_manager\")\r\ndef user_data(context, config):\r\n \"\"\"Load user events using Avro schema with validation.\"\"\"\r\n io_manager = context.resources.avro_kafka_io_manager\r\n return io_manager.load_input(\r\n context,\r\n topic=\"user-events\",\r\n schema_file=\"schemas/user.avsc\",\r\n validate_evolution=True\r\n )\r\n```\r\n\r\n### Protobuf Support\r\n\r\nHigh-performance binary format with full schema management.\r\n\r\n```python\r\nfrom dagster_kafka.protobuf_io_manager import create_protobuf_kafka_io_manager\r\n\r\nprotobuf_manager = create_protobuf_kafka_io_manager(\r\n kafka_resource=kafka_resource,\r\n schema_registry_url=\"http://localhost:8081\",\r\n consumer_group_id=\"protobuf-consumer\",\r\n enable_dlq=True\r\n)\r\n```\r\n\r\n## Confluent Connect Integration\r\n\r\nThe Confluent Connect integration provides a complete solution for managing Kafka Connect connectors within your Dagster environment.\r\n\r\n### Connect Management\r\n\r\n**REST API Client for Kafka Connect**\r\n\r\n```python\r\nfrom dagster_kafka.connect import ConfluentConnectClient\r\n\r\n# Create a client to interact with the Connect REST API\r\nclient = ConfluentConnectClient(base_url=\"http://localhost:8083\")\r\n\r\n# List available connectors\r\nconnectors = client.list_connectors()\r\n\r\n# Get connector status\r\nstatus = client.get_connector_status(\"my-connector\")\r\n\r\n# Create a new connector\r\nconnector_config = {\r\n \"name\": \"file-source-connector\",\r\n \"config\": {\r\n \"connector.class\": \"org.apache.kafka.connect.file.FileStreamSourceConnector\",\r\n \"tasks.max\": \"1\",\r\n \"file\": \"/tmp/test-source.txt\",\r\n \"topic\": \"test-topic\"\r\n }\r\n}\r\nclient.create_connector(connector_config)\r\n\r\n# Update configuration\r\nclient.update_connector_config(\"file-source-connector\", {\"tasks.max\": \"2\"})\r\n\r\n# Manage connector lifecycle\r\nclient.pause_connector(\"file-source-connector\")\r\nclient.resume_connector(\"file-source-connector\")\r\nclient.restart_connector(\"file-source-connector\")\r\n```\r\n\r\n### Connector Assets\r\n\r\n**Define Kafka Connect connectors as Dagster assets**\r\n\r\n```python\r\nfrom dagster import Definitions\r\nfrom dagster_kafka.connect import ConfluentConnectResource, create_connector_asset\r\n\r\n# Create a connector asset\r\nsource_connector = create_connector_asset(\r\n group_name=\"kafka_connect\",\r\n)\r\n\r\n# Define your Dagster job\r\ndefs = Definitions(\r\n assets=[source_connector],\r\n resources={\r\n \"connect\": ConfluentConnectResource(\r\n connect_url=\"http://localhost:8083\",\r\n )\r\n },\r\n)\r\n```\r\n\r\n#### Configuration Example\r\n\r\n```yaml\r\nops:\r\n connector_asset:\r\n config:\r\n name: \"mysql-source-connector\"\r\n connector_class: \"io.debezium.connector.mysql.MySqlConnector\"\r\n config:\r\n tasks.max: \"1\"\r\n database.hostname: \"mysql\"\r\n database.port: \"3306\"\r\n database.user: \"debezium\"\r\n database.password: \"dbz\"\r\n database.server.id: \"184054\"\r\n database.server.name: \"dbserver1\"\r\n database.include.list: \"inventory\"\r\n database.history.kafka.bootstrap.servers: \"kafka:9092\"\r\n database.history.kafka.topic: \"schema-changes.inventory\"\r\n```\r\n\r\n### Health Monitoring\r\n\r\n**Monitor connector health and implement auto-recovery**\r\n\r\n```python\r\nfrom dagster import job, op, sensor, SensorResult, RunRequest\r\nfrom dagster_kafka.connect import ConfluentConnectResource\r\n\r\n@sensor(\r\n name=\"connect_health_sensor\",\r\n minimum_interval_seconds=60,\r\n required_resource_keys={\"connect\"},\r\n)\r\ndef connector_health_sensor(context):\r\n \"\"\"Monitor the health of Kafka Connect connectors.\"\"\"\r\n connect = context.resources.connect\r\n connector_names = [\"mysql-source\", \"elasticsearch-sink\"]\r\n \r\n unhealthy_connectors = []\r\n \r\n # Check each connector\r\n for connector_name in connector_names:\r\n try:\r\n status = connect.get_connector_status(connector_name)\r\n connector_state = status[\"connector\"][\"state\"]\r\n \r\n # Check if connector is unhealthy\r\n if connector_state != \"RUNNING\":\r\n context.log.warning(\r\n f\"Connector {connector_name} is in {connector_state} state\"\r\n )\r\n unhealthy_connectors.append({\r\n \"name\": connector_name,\r\n \"state\": connector_state\r\n })\r\n \r\n # Also check tasks\r\n for task in status.get(\"tasks\", []):\r\n task_state = task.get(\"state\")\r\n task_id = task.get(\"id\")\r\n \r\n if task_state != \"RUNNING\":\r\n context.log.warning(\r\n f\"Task {task_id} of connector {connector_name} is in {task_state} state\"\r\n )\r\n unhealthy_connectors.append({\r\n \"name\": connector_name,\r\n \"task_id\": task_id,\r\n \"state\": task_state\r\n })\r\n except Exception as e:\r\n context.log.error(f\"Error checking connector {connector_name}: {e}\")\r\n \r\n # If we have unhealthy connectors, trigger a remediation job\r\n if unhealthy_connectors:\r\n return RunRequest(\r\n run_key=f\"connector_health_{context.cursor}\",\r\n job_name=\"remediate_connectors_job\",\r\n run_config={\r\n \"ops\": {\r\n \"remediate_connectors\": {\r\n \"config\": {\r\n \"unhealthy_connectors\": unhealthy_connectors\r\n }\r\n }\r\n }\r\n },\r\n )\r\n \r\n return SensorResult(\r\n skip_reason=\"All connectors healthy\",\r\n cursor=str(context.get_current_time()),\r\n )\r\n```\r\n\r\n### Recovery Patterns\r\n\r\n**Implement automated recovery from connector failures**\r\n\r\n```python\r\nfrom dagster_kafka.connect import ConfluentConnectClient\r\n\r\ndef advanced_recovery(client, connector_name, max_attempts=3):\r\n \"\"\"\r\n Advanced recovery pattern that tries different recovery methods\r\n based on the nature of the failure.\r\n \"\"\"\r\n for attempt in range(1, max_attempts + 1):\r\n try:\r\n # Check current status\r\n status = client.get_connector_status(connector_name)\r\n connector_state = status[\"connector\"][\"state\"]\r\n \r\n if connector_state == \"RUNNING\":\r\n # Check if any tasks are failing\r\n tasks = status.get(\"tasks\", [])\r\n failing_tasks = [\r\n task for task in tasks\r\n if task.get(\"state\") != \"RUNNING\"\r\n ]\r\n \r\n if not failing_tasks:\r\n return True\r\n \r\n # Restart only the failing tasks\r\n for task in failing_tasks:\r\n task_id = task.get(\"id\")\r\n client.restart_task(connector_name, int(task_id))\r\n else:\r\n # Try different recovery strategies based on state\r\n if connector_state == \"FAILED\":\r\n client.restart_connector(connector_name)\r\n elif connector_state == \"PAUSED\":\r\n client.resume_connector(connector_name)\r\n else:\r\n client.restart_connector(connector_name)\r\n client.resume_connector(connector_name)\r\n \r\n # Wait for recovery to take effect and check again\r\n time.sleep(5)\r\n status = client.get_connector_status(connector_name)\r\n if status[\"connector\"][\"state\"] == \"RUNNING\":\r\n return True\r\n except Exception as e:\r\n pass\r\n \r\n return False\r\n```\r\n\r\n### CLI Tools\r\n\r\nThe Confluent Connect integration includes several command-line tools for connector management and monitoring:\r\n\r\n```bash\r\n# List all connectors\r\npython connect_cli.py list\r\n\r\n# Get connector status\r\npython connect_cli.py status mysql-connector\r\n\r\n# Create a new connector\r\npython connect_cli.py create connector_config.json\r\n\r\n# Update connector configuration\r\npython connect_cli.py update mysql-connector updated_config.json\r\n\r\n# Restart a connector\r\npython connect_cli.py restart mysql-connector\r\n\r\n# Monitor connector health\r\npython connect_health_monitor.py --auto-restart mysql-connector elasticsearch-connector\r\n```\r\n\r\n## Enterprise Features\r\n\r\n### Comprehensive Enterprise Validation\r\n\r\n**Version 1.3.0** - Most validated Kafka integration package ever created:\r\n\r\n#### 12-Phase Enterprise Validation Completed\r\n- **EXCEPTIONAL Performance**: 1,199 messages/second peak throughput\r\n- **Security Hardened**: Complete credential validation + network security \r\n- **Stress Tested**: 100% success rate (305/305 operations over 8+ minutes)\r\n- **Memory Efficient**: Stable under extended load (+42MB over 8 minutes)\r\n- **Enterprise Ready**: Complete DLQ tooling suite with 5 CLI tools\r\n- **Zero Critical Issues**: Across all validation phases\r\n- **Connect Integration Validated**: Race condition and load testing complete\r\n- **JSON Schema Validated**: 4th serialization format thoroughly tested\r\n\r\n#### Validation Results Summary\r\n| Phase | Test Type | Result | Key Metrics |\r\n|-------|-----------|--------|-------------|\r\n| **Phase 5** | Performance Testing | \u2705 **PASS** | 1,199 msgs/sec peak throughput |\r\n| **Phase 7** | Integration Testing | \u2705 **PASS** | End-to-end message flow validated |\r\n| **Phase 9** | Compatibility Testing | \u2705 **PASS** | Python 3.12 + Dagster 1.11.3 |\r\n| **Phase 10** | Security Audit | \u2705 **PASS** | Credential + network security |\r\n| **Phase 11** | Stress Testing | \u2705 **EXCEPTIONAL** | 100% success rate, 305 operations |\r\n| **Phase 12** | Connect Integration | \u2705 **PASS** | Race condition and load testing |\r\n\r\n### Core Enterprise Features\r\n- **Complete Security**: SASL/SSL authentication and encryption\r\n- **Schema Evolution**: Breaking change detection across all formats\r\n- **Production Monitoring**: Real-time alerting with Slack/Email integration\r\n- **High Performance**: Advanced caching, batching, and connection pooling\r\n- **Error Recovery**: Multiple recovery strategies for production resilience\r\n- **Dagster Components**: YAML-based configuration for teams\r\n- **Connect Integration**: Complete Kafka Connect management\r\n\r\n### Enterprise DLQ Tooling Suite\r\n\r\nComplete operational tooling for Dead Letter Queue management:\r\n\r\n```bash\r\n# Analyze failed messages with comprehensive error pattern analysis\r\ndlq-inspector --topic user-events --max-messages 20\r\n\r\n# Replay messages with filtering and safety controls \r\ndlq-replayer --source-topic orders_dlq --target-topic orders --dry-run\r\n\r\n# Monitor DLQ health across multiple topics\r\ndlq-monitor --topics user-events_dlq,orders_dlq --output-format json\r\n\r\n# Set up automated alerting\r\ndlq-alerts --topic critical-events_dlq --max-messages 500\r\n\r\n# Operations dashboard for DLQ health monitoring\r\ndlq-dashboard --topics user-events_dlq,orders_dlq\r\n```\r\n\r\n## Dead Letter Queue (DLQ)\r\n\r\n### DLQ Strategies\r\n- **DISABLED**: No DLQ processing\r\n- **IMMEDIATE**: Send to DLQ immediately on failure\r\n- **RETRY_THEN_DLQ**: Retry N times, then send to DLQ\r\n- **CIRCUIT_BREAKER**: Circuit breaker pattern with DLQ fallback\r\n\r\n### Error Classification\r\n- **DESERIALIZATION_ERROR**: Failed to deserialize message\r\n- **SCHEMA_ERROR**: Schema validation failed (includes JSON Schema validation)\r\n- **PROCESSING_ERROR**: Business logic error\r\n- **CONNECTION_ERROR**: Kafka connection issues\r\n- **TIMEOUT_ERROR**: Message processing timeout\r\n- **UNKNOWN_ERROR**: Unclassified errors\r\n\r\n### Circuit Breaker Configuration\r\n```python\r\nfrom dagster_kafka import DLQConfiguration, DLQStrategy\r\n\r\ndlq_config = DLQConfiguration(\r\n strategy=DLQStrategy.CIRCUIT_BREAKER,\r\n circuit_breaker_failure_threshold=5, # Open after 5 failures\r\n circuit_breaker_recovery_timeout_ms=30000, # Test recovery after 30s\r\n circuit_breaker_success_threshold=2 # Close after 2 successes\r\n)\r\n```\r\n\r\n## Security\r\n\r\n### Security Protocols Supported\r\n- **PLAINTEXT**: For local development and testing\r\n- **SSL**: Certificate-based encryption\r\n- **SASL_PLAINTEXT**: Username/password authentication \r\n- **SASL_SSL**: Combined authentication and encryption (recommended for production)\r\n\r\n### SASL Authentication Mechanisms\r\n- **PLAIN**: Simple username/password authentication\r\n- **SCRAM-SHA-256**: Secure challenge-response authentication\r\n- **SCRAM-SHA-512**: Enhanced secure authentication\r\n- **GSSAPI**: Kerberos authentication for enterprise environments\r\n- **OAUTHBEARER**: OAuth-based authentication\r\n\r\n### Secure Production Example\r\n```python\r\nfrom dagster_kafka import KafkaResource, SecurityProtocol, SaslMechanism\r\n\r\nsecure_kafka = KafkaResource(\r\n bootstrap_servers=\"prod-kafka-01:9092,prod-kafka-02:9092\",\r\n security_protocol=SecurityProtocol.SASL_SSL,\r\n sasl_mechanism=SaslMechanism.SCRAM_SHA_256,\r\n sasl_username=\"production-user\",\r\n sasl_password=\"secure-password\",\r\n ssl_ca_location=\"/etc/ssl/certs/kafka-ca.pem\",\r\n ssl_check_hostname=True\r\n)\r\n```\r\n\r\n## Performance\r\n\r\n### Validated Performance Results\r\n- **Peak Throughput**: 1,199 messages/second\r\n- **Stress Test Success**: 100% (305/305 operations)\r\n- **Extended Stability**: 8+ minutes continuous operation\r\n- **Memory Efficiency**: +42MB over extended load (excellent)\r\n- **Concurrent Operations**: 120/120 successful operations\r\n- **Resource Management**: Zero thread accumulation\r\n\r\n### Enterprise Stability Testing\r\n```\r\nPASS Extended Stability: 5+ minutes, 137/137 successful materializations\r\nPASS Resource Management: 15 cycles, no memory leaks detected \r\nPASS Concurrent Usage: 8 threads \u00d7 15 operations = 100% success\r\nPASS Comprehensive Stress: 8+ minutes, 305 operations, EXCEPTIONAL rating\r\n```\r\n\r\n## Examples\r\n\r\n### Complete Kafka Ecosystem Integration\r\n\r\n```python\r\nfrom dagster import Definitions, asset, AssetExecutionContext\r\nfrom dagster_kafka import KafkaResource\r\nfrom dagster_kafka.connect import ConfluentConnectResource, create_connector_asset\r\nfrom dagster_kafka.json_schema_io_manager import create_json_schema_kafka_io_manager\r\n\r\n# Source connector that ingests data into Kafka\r\nsource_connector = create_connector_asset(\r\n key_prefix=[\"mysql\", \"cdc\"],\r\n group_name=\"source_connectors\",\r\n)\r\n\r\n# Processing asset that consumes and transforms data\r\n@asset(\r\n key_prefix=[\"kafka\", \"processed\"],\r\n group_name=\"processing\",\r\n deps=[source_connector], # Depends on source connector\r\n io_manager_key=\"json_schema_io_manager\",\r\n)\r\ndef transform_data(context: AssetExecutionContext):\r\n \"\"\"Transform the data from the source topic.\"\"\"\r\n # Your transformation logic\r\n return {\"transformed\": \"data\"}\r\n\r\n# Sink connector that exports data to external systems\r\nsink_connector = create_connector_asset(\r\n key_prefix=[\"elasticsearch\", \"sink\"],\r\n group_name=\"sink_connectors\",\r\n deps=[transform_data], # Depends on the transformation\r\n)\r\n\r\n# Define your complete pipeline with Kafka ecosystem integration\r\ndefs = Definitions(\r\n assets=[source_connector, transform_data, sink_connector],\r\n resources={\r\n # Kafka resource for message consumption/production\r\n \"kafka\": KafkaResource(\r\n bootstrap_servers=\"localhost:9092\",\r\n ),\r\n \r\n # Connect resource for connector management\r\n \"connect\": ConfluentConnectResource(\r\n connect_url=\"http://localhost:8083\",\r\n ),\r\n \r\n # JSON Schema validation for data quality\r\n \"json_schema_io_manager\": create_json_schema_kafka_io_manager(\r\n kafka_resource=KafkaResource(bootstrap_servers=\"localhost:9092\"),\r\n schema_file=\"schemas/event_schema.json\",\r\n enable_schema_validation=True,\r\n strict_validation=True,\r\n ),\r\n },\r\n)\r\n```\r\n\r\n### Connector Health Monitoring Integration\r\n\r\n```python\r\nfrom dagster import Definitions, job, op, sensor, Config\r\nfrom dagster_kafka.connect import ConfluentConnectResource\r\n\r\nclass UnhealthyConnectorsConfig(Config):\r\n unhealthy_connectors: List[Dict[str, Any]]\r\n\r\n@op\r\ndef remediate_connectors(context, config: UnhealthyConnectorsConfig):\r\n \"\"\"Auto-remediate unhealthy connectors.\"\"\"\r\n connect = context.resources.connect\r\n unhealthy_connectors = config.unhealthy_connectors\r\n \r\n for connector in unhealthy_connectors:\r\n connector_name = connector[\"name\"]\r\n \r\n # If it's a task issue, restart the specific task\r\n if \"task_id\" in connector:\r\n task_id = connector[\"task_id\"]\r\n context.log.info(f\"Restarting task {task_id} of connector {connector_name}\")\r\n connect.restart_task(connector_name, int(task_id))\r\n \r\n # Otherwise restart the entire connector\r\n else:\r\n context.log.info(f\"Restarting connector {connector_name}\")\r\n connect.restart_connector(connector_name)\r\n \r\n # Resume if it was paused\r\n context.log.info(f\"Resuming connector {connector_name}\")\r\n connect.resume_connector(connector_name)\r\n \r\n return {\r\n \"remediated_count\": len(unhealthy_connectors),\r\n \"connector_names\": [c[\"name\"] for c in unhealthy_connectors]\r\n }\r\n\r\n@job\r\ndef remediate_connectors_job():\r\n \"\"\"Job to remediate unhealthy connectors.\"\"\"\r\n remediate_connectors()\r\n\r\n# Define your Dagster components\r\ndefs = Definitions(\r\n jobs=[remediate_connectors_job],\r\n sensors=[connector_health_sensor], # Use the sensor defined earlier\r\n resources={\r\n \"connect\": ConfluentConnectResource(\r\n connect_url=\"http://localhost:8083\",\r\n )\r\n },\r\n)\r\n```\r\n\r\n## Development\r\n\r\n### Running Tests\r\n```bash\r\n# Run all validation tests (12 phases)\r\npython -m pytest tests/ -v\r\n\r\n# Specific test modules\r\npython -m pytest tests/test_connect_client.py -v # Connect client tests\r\npython -m pytest tests/test_connect_resource.py -v # Connect resource tests\r\npython -m pytest tests/test_connect_assets.py -v # Connect assets tests \r\npython -m pytest tests/test_json_schema_io_manager.py -v # JSON Schema tests\r\npython -m pytest tests/test_avro_io_manager.py -v # Avro tests\r\npython -m pytest tests/test_protobuf_io_manager.py -v # Protobuf tests\r\npython -m pytest tests/test_dlq.py -v # DLQ tests\r\npython -m pytest tests/test_security.py -v # Security tests\r\npython -m pytest tests/test_performance.py -v # Performance tests\r\n```\r\n\r\n### Local Development Setup\r\n```bash\r\n# Clone the repository\r\ngit clone https://github.com/kingsley-123/dagster-kafka-integration.git\r\ncd dagster-kafka-integration\r\n\r\n# Install dependencies\r\npip install -r requirements.txt\r\n\r\n# Install in development mode\r\npip install -e .\r\n\r\n# Start local Kafka and Connect for testing\r\ndocker-compose up -d\r\n```\r\n\r\n### Example Directory Structure\r\n```\r\nexamples/\r\n\u251c\u2500\u2500 json_examples/ # Basic JSON message examples\r\n\u251c\u2500\u2500 json_schema_examples/ # JSON Schema validation examples\r\n\u251c\u2500\u2500 avro_examples/ # Avro schema examples \r\n\u251c\u2500\u2500 protobuf_examples/ # Protobuf examples\r\n\u251c\u2500\u2500 connect_examples/ # Confluent Connect integration examples (NEW)\r\n\u251c\u2500\u2500 components_examples/ # YAML Components configuration\r\n\u251c\u2500\u2500 dlq_examples/ # Complete DLQ tooling suite\r\n\u251c\u2500\u2500 security_examples/ # Enterprise security examples\r\n\u251c\u2500\u2500 performance_examples/ # Performance optimization\r\n\u2514\u2500\u2500 production_examples/ # Enterprise deployment patterns\r\n```\r\n\r\n## Why Choose This Integration\r\n\r\n### Complete Solution\r\n- **Only integration supporting all 4 major formats** (JSON, JSON Schema, Avro, Protobuf)\r\n- **Complete Kafka ecosystem integration** with Confluent Connect support\r\n- **Enterprise-grade security** with SASL/SSL support\r\n- **Production-ready** with comprehensive monitoring\r\n- **Advanced error handling** with Dead Letter Queue support\r\n- **Complete DLQ Tooling Suite** for enterprise operations\r\n\r\n### Developer Experience\r\n- **Multiple configuration options** - Python API OR simple YAML Components\r\n- **Team accessibility** - Components enable non-Python users to configure assets\r\n- **Familiar Dagster patterns** - feels native to the platform\r\n- **Comprehensive examples** for all use cases including security and DLQ\r\n- **Extensive documentation** and testing\r\n- **Production-ready CLI tooling** for DLQ management and Connect operations\r\n\r\n### Enterprise Ready\r\n- **12-phase comprehensive validation** covering all scenarios\r\n- **Real-world deployment patterns** and examples\r\n- **Performance optimization** tools and monitoring\r\n- **Enterprise security** for production Kafka clusters\r\n- **Bulletproof error handling** with circuit breaker patterns\r\n- **Complete operational tooling** for DLQ management and Connect integration\r\n\r\n### Unprecedented Validation\r\n- **Most validated package** in the Dagster ecosystem\r\n- **Performance proven**: 1,199 msgs/sec peak throughput\r\n- **Stability proven**: 100% success rate under stress\r\n- **Security proven**: Complete credential and network validation\r\n- **Enterprise proven**: Exceptional rating across all dimensions\r\n\r\n## Roadmap\r\n\r\n### Completed Features (v1.3.0)\r\n- **JSON Support** - Complete native integration \u2705\r\n- **JSON Schema Support** - Data validation with evolution checking \u2705\r\n- **Avro Support** - Full Schema Registry + evolution validation \u2705\r\n- **Protobuf Support** - Complete Protocol Buffers integration \u2705\r\n- **Confluent Connect Integration** - Complete Connect REST API integration \u2705\r\n- **Connector Assets** - Define connectors as Dagster assets \u2705\r\n- **Health Monitoring** - Automated connector health monitoring \u2705\r\n- **Recovery Patterns** - Advanced connector recovery strategies \u2705\r\n- **Connect CLI Tools** - Complete command-line tools for Connect operations \u2705\r\n- **Dagster Components** - YAML-based configuration support \u2705\r\n- **Enterprise Security** - Complete SASL/SSL authentication and encryption \u2705\r\n- **Schema Evolution** - All compatibility levels across formats \u2705\r\n- **Production Monitoring** - Real-time alerting and metrics \u2705\r\n- **High-Performance Optimization** - Caching, batching, pooling \u2705\r\n- **Dead Letter Queues** - Advanced error handling with circuit breaker \u2705\r\n- **Complete DLQ Tooling Suite** - Inspector, Replayer, Monitoring, Alerting \u2705\r\n- **Comprehensive Testing** - 12-phase enterprise validation \u2705\r\n- **PyPI Distribution** - Official package published and validated \u2705\r\n- **Security Hardening** - Configuration injection protection \u2705\r\n\r\n### Upcoming Features\r\n- **Enhanced JSON Schema** - Schema registry integration\r\n- **Advanced Connect Monitoring** - Custom metrics and dashboards\r\n- **Connect Templates** - Pre-built connector configurations\r\n\r\n## Contributing\r\n\r\nContributions are welcome! This project aims to be the definitive Kafka integration for Dagster.\r\n\r\n### Ways to contribute:\r\n- **Report issues** - Found a bug? Let us know!\r\n- **Feature requests** - What would make this more useful?\r\n- **Documentation** - Help improve examples and guides\r\n- **Code contributions** - PRs welcome for any improvements\r\n- **Security testing** - Help test security configurations\r\n- **DLQ testing** - Help test error handling scenarios\r\n- **Connect testing** - Help test connector integration scenarios\r\n\r\n## License\r\n\r\nApache 2.0 - see [LICENSE](LICENSE) file for details.\r\n\r\n## Community & Support\r\n\r\n- **GitHub Issues**: [Report bugs and request features](https://github.com/kingsley-123/dagster-kafka-integration/issues)\r\n- **GitHub Discussions**: [Share use cases and get help](https://github.com/kingsley-123/dagster-kafka-integration/discussions)\r\n- **PyPI Package**: [Install and documentation](https://pypi.org/project/dagster-kafka/)\r\n- **Star the repo**: If this helped your project!\r\n\r\n## Acknowledgments\r\n\r\n- **Dagster Community**: For the initial feature request and continued feedback\r\n- **Contributors**: Thanks to all who provided feedback, testing, and code contributions\r\n- **Enterprise Users**: Built in response to real production deployment needs\r\n- **Security Community**: Special thanks for security testing and validation\r\n- **JSON Schema Community**: Thanks for validation methodology and best practices\r\n- **Confluent Community**: For guidance on Connect integration best practices\r\n\r\n---\r\n\r\n## The Complete Enterprise Solution\r\n\r\n**The most comprehensively validated Kafka integration for Dagster** - supporting all four major serialization formats (JSON, JSON Schema, Avro, Protobuf) with enterprise-grade production features, complete security, advanced Dead Letter Queue error handling, Confluent Connect integration, YAML-based Components, and complete operational tooling suite.\r\n\r\n**Version 1.3.1** - Confluent Connect Integration Release\r\n\r\n*Built by [Kingsley Okonkwo](https://github.com/kingsley-123) - Solving real data engineering problems with comprehensive open source solutions.*\r\n</artifact>\r\n</artifacts>\r\n\r\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Enterprise-grade Kafka integration for Dagster with Confluent Connect, comprehensive serialization support, DLQ handling, and production monitoring",
"version": "1.3.1",
"project_urls": {
"Bug Reports": "https://github.com/kingsley-123/dagster-kafka-integration/issues",
"Documentation": "https://github.com/kingsley-123/dagster-kafka-integration/blob/main/README.md",
"Homepage": "https://github.com/kingsley-123/dagster-kafka-integration",
"PyPI": "https://pypi.org/project/dagster-kafka/",
"Repository": "https://github.com/kingsley-123/dagster-kafka-integration",
"Source": "https://github.com/kingsley-123/dagster-kafka-integration"
},
"split_keywords": [
"dagster",
" kafka",
" apache-kafka",
" streaming",
" data-engineering",
" data-pipeline",
" etl",
" data-processing",
" json",
" json-schema",
" avro",
" protobuf",
" serialization",
" enterprise",
" production",
" monitoring",
" alerting",
" dlq",
" dead-letter-queue",
" error-handling",
" circuit-breaker",
" microservices",
" distributed-systems",
" real-time",
" schema-registry",
" confluent",
" data-validation",
" sasl",
" ssl",
" security",
" authentication",
" authorization",
" confluent-connect",
" kafka-connect",
" connectors",
" cdc"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "26c5c2608fb78af4187bada69c2aab35d2661ce03c31b8f10f83a66b4dbf0fe5",
"md5": "fff7e5355be26cf5b8d9be0c71cce205",
"sha256": "6d68a7041c6cd7cae9d50188a76751d70539dcf72bc5cf8f34aa188a7e00146a"
},
"downloads": -1,
"filename": "dagster_kafka-1.3.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "fff7e5355be26cf5b8d9be0c71cce205",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 83057,
"upload_time": "2025-08-21T21:48:42",
"upload_time_iso_8601": "2025-08-21T21:48:42.023436Z",
"url": "https://files.pythonhosted.org/packages/26/c5/c2608fb78af4187bada69c2aab35d2661ce03c31b8f10f83a66b4dbf0fe5/dagster_kafka-1.3.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "6a8227caef0264da7ae7246c91142c647ad028e7e5034c1b9c6000bbbb3432a0",
"md5": "4fe1797d1f41956143e1acc401dfd140",
"sha256": "0d189c9a6041bb2ba86dc3159da927e585a66997277da064c8418159c13d4b8a"
},
"downloads": -1,
"filename": "dagster_kafka-1.3.1.tar.gz",
"has_sig": false,
"md5_digest": "4fe1797d1f41956143e1acc401dfd140",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 101835,
"upload_time": "2025-08-21T21:48:43",
"upload_time_iso_8601": "2025-08-21T21:48:43.699206Z",
"url": "https://files.pythonhosted.org/packages/6a/82/27caef0264da7ae7246c91142c647ad028e7e5034c1b9c6000bbbb3432a0/dagster_kafka-1.3.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-21 21:48:43",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "kingsley-123",
"github_project": "dagster-kafka-integration",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [],
"lcname": "dagster-kafka"
}
JSON
