from confluent_kafka import Producer

producer = Producer({
    'bootstrap.servers': 'kafka-1:9092,kafka-2:9092,kafka-3:9092',
    
    # Reliability settings
    'acks': 'all',                    # Wait for all replicas
    'enable.idempotence': True,       # Exactly-once delivery
    'max.in.flight.requests.per.connection': 5,  # Required for idempotence
    
    # Performance settings
    'linger.ms': 5,                   # Batch for up to 5ms
    'batch.size': 65536,              # 64KB batches
    'compression.type': 'snappy',     # Good compression/CPU tradeoff
    
    # Retry settings
    'retries': 2147483647,            # Retry forever
    'retry.backoff.ms': 100,
    'delivery.timeout.ms': 120000,    # Give up after 2 minutes
})

def delivery_report(err, msg):
    if err is not None:
        print(f'Message delivery failed: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}] @ {msg.offset()}')

# Produce with structured key for partitioning
producer.produce(
    topic='user-events',
    key=user_id.encode('utf-8'),     # Ensures same user → same partition
    value=json.dumps(event).encode('utf-8'),
    headers={'event-type': event['type'], 'schema-version': '2'},
    on_delivery=delivery_report
)
producer.poll(0)

from confluent_kafka.avro import AvroProducer
from confluent_kafka import avro

USER_CLICK_SCHEMA = avro.loads("""
{
  "type": "record",
  "name": "UserClick",
  "namespace": "com.myapp.events",
  "fields": [
    {"name": "user_id", "type": "string"},
    {"name": "session_id", "type": "string"},
    {"name": "page_url", "type": "string"},
    {"name": "timestamp_ms", "type": "long"},
    {"name": "referrer", "type": ["null", "string"], "default": null}
  ]
}
""")

producer = AvroProducer(
    {
        'bootstrap.servers': 'kafka-1:9092',
        'schema.registry.url': 'http://schema-registry:8081'
    },
    default_value_schema=USER_CLICK_SCHEMA
)

producer.produce(
    topic='user-clicks',
    key=user_id,
    value={
        'user_id': user_id,
        'session_id': session_id,
        'page_url': '/products/widget-123',
        'timestamp_ms': int(time.time() * 1000),
        'referrer': None
    }
)

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaSink
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common import WatermarkStrategy
from pyflink.datastream.window import TumblingEventTimeWindows, Time

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)
env.enable_checkpointing(60000)   # Checkpoint every 60s

# Kafka source with watermark strategy
source = (KafkaSource.builder()
    .set_bootstrap_servers("kafka-1:9092,kafka-2:9092")
    .set_topics("user-clicks")
    .set_group_id("flink-click-aggregator")
    .set_starting_offsets(KafkaOffsetsInitializer.committed_offsets())
    .set_value_only_deserializer(SimpleStringSchema())
    .build())

watermark_strategy = (WatermarkStrategy
    .for_bounded_out_of_orderness(Duration.of_seconds(10))
    .with_timestamp_assigner(ClickEventTimestampAssigner()))

stream = env.from_source(
    source=source,
    watermark_strategy=watermark_strategy,
    source_name="Kafka Click Events"
)

from pyflink.datastream.functions import ProcessWindowFunction
from pyflink.datastream import KeyedStream

class SessionAnalyzer(ProcessWindowFunction):
    """Analyze user sessions: duration, pages, conversion."""
    
    def process(self, key, context, elements):
        clicks = sorted(elements, key=lambda e: e.timestamp_ms)
        
        if not clicks:
            return
        
        session_duration_ms = clicks[-1].timestamp_ms - clicks[0].timestamp_ms
        pages_visited = len(set(c.page_url for c in clicks))
        converted = any(c.page_url.startswith('/checkout') for c in clicks)
        
        yield SessionSummary(
            user_id=key,
            session_start=clicks[0].timestamp_ms,
            session_end=clicks[-1].timestamp_ms,
            duration_ms=session_duration_ms,
            page_count=pages_visited,
            click_count=len(clicks),
            converted=converted
        )

# Session window: gap of 30 minutes = new session
session_results = (stream
    .map(parse_click_event)
    .key_by(lambda e: e.user_id)
    .window(EventTimeSessionWindows.with_gap(Time.minutes(30)))
    .process(SessionAnalyzer()))

from pyflink.datastream.connectors.kafka import KafkaSink, KafkaRecordSerializationSchema
from pyflink.datastream.connectors.kafka import DeliveryGuarantee

# Flink Kafka sink with exactly-once semantics
sink = (KafkaSink.builder()
    .set_bootstrap_servers("kafka-1:9092,kafka-2:9092")
    .set_record_serializer(
        KafkaRecordSerializationSchema.builder()
            .set_topic("session-summaries")
            .set_value_serialization_schema(SessionSummarySchema())
            .build()
    )
    .set_delivery_guarantee(DeliveryGuarantee.EXACTLY_ONCE)  # 👈 Key setting
    .set_transactional_id_prefix("flink-session-analyzer")
    .build())

session_results.sink_to(sink)

# Checkpoint configuration for exactly-once
env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)
env.get_checkpoint_config().set_min_pause_between_checkpoints(30000)
env.get_checkpoint_config().set_checkpoint_timeout(120000)

from pyflink.table import StreamTableEnvironment, EnvironmentSettings

settings = EnvironmentSettings.new_instance().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(stream_env, environment_settings=settings)

# Register Kafka CDC table (orders)
t_env.execute_sql("""
CREATE TABLE orders (
  order_id BIGINT,
  user_id STRING,
  product_id STRING,
  amount DECIMAL(10,2),
  status STRING,
  created_at TIMESTAMP(3),
  op STRING,  -- CDC operation: 'c', 'u', 'd', 'r'
  WATERMARK FOR created_at AS created_at - INTERVAL '10' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'prod-postgres.public.orders',
  'properties.bootstrap.servers' = 'kafka-1:9092',
  'properties.group.id' = 'flink-order-processor',
  'format' = 'debezium-avro-confluent',
  'debezium-avro-confluent.schema-registry.url' = 'http://schema-registry:8081'
)
""")

# Real-time revenue per user (1-hour tumbling window)
result = t_env.sql_query("""
SELECT
  user_id,
  TUMBLE_START(created_at, INTERVAL '1' HOUR) AS window_start,
  COUNT(*) AS order_count,
  SUM(amount) AS total_revenue
FROM orders
WHERE status = 'completed' AND op IN ('c', 'r')
GROUP BY user_id, TUMBLE(created_at, INTERVAL '1' HOUR)
""")

# In your Flink job, register custom metrics
from pyflink.datastream.functions import RichMapFunction

class EnrichedProcessor(RichMapFunction):
    def open(self, runtime_context):
        # Register metrics
        self.events_processed = runtime_context \
            .get_metrics_group() \
            .counter("events_processed")
        self.enrichment_latency = runtime_context \
            .get_metrics_group() \
            .histogram("enrichment_latency_ms", DropwizardHistogramWrapper(
                ExponentiallyDecayingReservoir()
            ))
    
    def map(self, event):
        start = time.time()
        enriched = enrich_event(event)
        
        self.events_processed.inc()
        self.enrichment_latency.update(int((time.time() - start) * 1000))
        
        return enriched

# WRONG: Auto-commit can lose messages on crash
consumer = KafkaConsumer(
    bootstrap_servers=['kafka-1:9092'],
    enable_auto_commit=True  # ← dangerous
)

# RIGHT: Manual commit after processing
consumer = KafkaConsumer(
    bootstrap_servers=['kafka-1:9092'],
    enable_auto_commit=False,
    max_poll_records=500
)

for message_batch in consumer:
    process_batch(message_batch)
    consumer.commit()  # Only commit after successful processing

stream.key_by(lambda e: e.user_id) \
    .window(TumblingEventTimeWindows.of(Time.minutes(5))) \
    .allowed_lateness(Time.minutes(2)) \     # ← Accept up to 2 min late
    .side_output_late_data(late_events_tag) \  # ← Route to side output
    .aggregate(ClickCountAggregator())