// 1. Bounded out-of-orderness (most common production choice)
WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner((e, ts) -> e.getEventTs())
// Watermark = max(seen_event_ts) - 5s
// "I'll wait 5 seconds for stragglers, then fire windows"

// 2. Monotonically increasing (data is perfectly ordered)
WatermarkStrategy
    .<Event>forMonotonousTimestamps()
    .withTimestampAssigner((e, ts) -> e.getEventTs())
// Watermark = max(seen_ts) - 1ms. Zero delay. Use only if certain.

// 3. Custom — per-partition watermarks (Kafka-native)
WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(30))
    .withTimestampAssigner(...)
    .withIdleness(Duration.ofMinutes(5))
// withIdleness: if a Kafka partition has no events for 5min,
// don't let it block the global watermark from advancing.
// Critical for low-volume topics with sparse partitions.

// 4. Custom WatermarkGenerator (full control)
public class OrderedEventTimeGenerator
        implements WatermarkGenerator<Event> {
    private long maxTs = Long.MIN_VALUE;

    public void onEvent(Event e, long ts, WatermarkOutput out) {
        maxTs = Math.max(maxTs, e.getTs());
    }
    public void onPeriodicEmit(WatermarkOutput out) {
        out.emitWatermark(new Watermark(maxTs - 5000));
    }
}

OutputTag<Order> lateTag = new OutputTag<>("late-orders"){};

SingleOutputStreamOperator<Revenue> result = orders
    .keyBy(o -> o.customerId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))

    // Strategy 1: Allow late events to re-fire the window
    // Window fires at watermark, then AGAIN for each late event
    // for up to 'allowedLateness' past the window close
    .allowedLateness(Time.minutes(5))

    // Strategy 2: Route very-late events to a side output
    // (events arriving after allowedLateness period)
    .sideOutputLateData(lateTag)

    .aggregate(new RevenueAgg());

// Strategy 3: Process side output separately (correction stream)
DataStream<Order> lateOrders = result.getSideOutput(lateTag);
lateOrders.sinkTo(lateOrdersAuditTopic);

/*
 TIMELINE:
  t=10:00:00  window [09:00, 10:00) opens
  t=10:00:05  watermark passes 10:00 → window FIRES (main output)
  t=10:02:00  late event arrives (ts=09:58) → allowedLateness=5min
              still within window → fires AGAIN with updated result
  t=10:05:01  another late event (ts=09:45) → past allowedLateness
              routed to lateTag side output
*/

// ALL keyed state is scoped to: (operator, key, namespace)
// Flink manages serialization, checkpointing, and TTL.

// 1. ValueState — single value per key
ValueState<Long> count = ctx.getState(
    new ValueStateDescriptor<>("event-count", Long.class));
count.update(count.value() + 1);
Long v = count.value();
count.clear();

// 2. ListState — append-only list per key
ListState<Event> buffer = ctx.getListState(
    new ListStateDescriptor<>("event-buffer", Event.class));
buffer.add(event);
Iterable<Event> all = buffer.get();
buffer.update(newList);   // replace all

// 3. MapState — key-value map per Flink-key
MapState<String, Double> totals = ctx.getMapState(
    new MapStateDescriptor<>("category-totals",
        String.class, Double.class));
totals.put("electronics", totals.get("electronics") + amount);

// 4. ReducingState — auto-reduces on add()
ReducingState<Long> sum = ctx.getReducingState(
    new ReducingStateDescriptor<>("sum", Long::sum, Long.class));
sum.add(42L);   // calls (current, 42L) → Long::sum
sum.get();       // current aggregate

// 5. AggregatingState — complex incremental aggregation
AggregatingState<Order, Stats> statsState = ctx.getAggregatingState(
    new AggregatingStateDescriptor<>("stats",
        new StatsAggregator(), Stats.class));

// 6. State TTL — automatic expiry (prevents unbounded growth)
StateTtlConfig ttl = StateTtlConfig
    .newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter(1000)  // compact during RocksDB flush
    .build();
valueStateDescriptor.enableTimeToLive(ttl);

CheckpointConfig cfg = env.getCheckpointConfig();
cfg.setCheckpointInterval(60_000);       // every 60s
cfg.setCheckpointTimeout(120_000);       // fail if > 2min
cfg.setMinPauseBetweenCheckpoints(30_000); // cooldown
cfg.setMaxConcurrentCheckpoints(1);       // 1 in-flight at a time
cfg.setExternalizedCheckpointCleanup(      // keep on cancel
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

// Incremental checkpoints (RocksDB only — huge for large state):
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));
// true = incremental. Only changed SSTable files uploaded.
// Full checkpoint: 100GB → 30min. Incremental: 100GB → 90s.

// Savepoint: operator-initiated, used for upgrades
// $ flink savepoint <jobId> s3://bucket/savepoints/
// $ flink run -s s3://bucket/savepoints/sp-123 job.jar
// Savepoints survive code changes (with compatible state schema)

// 1. TUMBLING EVENT TIME — fixed, non-overlapping
stream.keyBy(k -> k.id)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .aggregate(new RevenueAgg());
// [00:00,01:00), [01:00,02:00) ... each fires once

// 2. SLIDING — overlapping windows
stream.keyBy(k -> k.id)
    .window(SlidingEventTimeWindows.of(
        Time.minutes(10),   // window size
        Time.minutes(5)))   // slide interval
    .aggregate(new RevenueAgg());
// Each event belongs to size/slide = 2 windows.
// [00:00,00:10), [00:05,00:15), [00:10,00:20) ...

// 3. SESSION — activity gap based, per key
stream.keyBy(k -> k.userId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
    .aggregate(new SessionAgg());
// Window extends as long as events keep arriving within 30min.
// No events for 30min → window closes → session emitted.

// 4. GLOBAL WINDOW — one window per key, forever
stream.keyBy(k -> k.id)
    .window(GlobalWindows.create())
    .trigger(new CountTrigger(100))    // fire every 100 events
    .aggregate(new RevenueAgg());
// Global window never closes without a custom trigger.
// Use for: arbitrary triggers (ML model inference every N records)

// 5. CUSTOM TRIGGER — fire on condition
stream.keyBy(k -> k.customerId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .trigger(PurgingTrigger.of(CountTrigger.of(50)))
    // Fire AND PURGE when 50 events accumulate within the window.
    // Fire again at window end. Useful for early/speculative results.
    .aggregate(new RevenueAgg());

/**
 * AggregateFunction: INCREMENTAL — O(1) state per window.
 * Accumulator updated per event. Efficient but no access
 * to window boundaries or all events at once.
 */
public class RevenueAgg
        implements AggregateFunction<Order, Double, Revenue> {

    public Double createAccumulator() { return 0.0; }
    public Double add(Order o, Double acc) { return acc + o.amount; }
    public Double merge(Double a, Double b) { return a + b; }
    public Revenue getResult(Double acc) {
        return new Revenue(acc);
    }
}

/**
 * ProcessWindowFunction: FULL — buffers ALL events in window.
 * Expensive (O(n) state) but gives access to window bounds,
 * all events, and iteration. Use for: percentiles, medians,
 * complex multi-pass calculations.
 */
public class P99Calculator
        extends ProcessWindowFunction<Event, Metrics, String, TimeWindow> {

    public void process(String key,
                         Context ctx,
                         Iterable<Event> events,   // ALL events
                         Collector<Metrics> out) {

        List<Long> latencies = StreamSupport
            .stream(events.spliterator(), false)
            .map(Event::getLatency)
            .sorted()
            .collect(Collectors.toList());

        long p99 = latencies.get((int)(latencies.size() * 0.99));
        out.collect(new Metrics(key,
            ctx.window().getStart(), ctx.window().getEnd(), p99));
    }
}

// BEST PRACTICE: combine both — incremental + window context:
stream.keyBy(...)
    .window(...)
    .aggregate(
        new RevenueAgg(),    // incremental: O(1) state
        new EnrichWithWindowBounds()  // called once on fire
    );

/*
 TWO-PHASE COMMIT SEQUENCE:
 ─────────────────────────────────────────────────

 PHASE 1 — PRE-COMMIT (on checkpoint barrier):
   1. Source operators snapshot their Kafka offsets
   2. Processing operators snapshot their keyed state
   3. Kafka sink begins a Kafka transaction
      (transactional.id = "flink-sink-{subtaskId}-{checkpointId}")
   4. All records written so far go into the open transaction
   5. Sink snapshots: transactional.id into checkpoint state
   6. All operators report checkpoint complete to JobManager

 PHASE 2 — COMMIT (after all operators confirm):
   1. JobManager receives ALL confirmements → checkpoint complete
   2. Kafka sink commits the open transaction (commit_transaction())
   3. Records NOW visible to read_committed consumers

 FAILURE SCENARIOS:
   ● Crash before phase 2: transaction aborted, no duplicates
     → restart from last complete checkpoint, reprocess from offsets
   ● Crash mid phase 2: transactional.id persisted in checkpoint
     → on restart, sink recovers transactional.id, aborts/commits
     → idempotent: committing an already-committed transaction is a no-op
*/

KafkaSink<Revenue> kafkaSink = KafkaSink.<Revenue>builder()
    .setBootstrapServers("broker:9092")
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("hourly-revenue-sink")
    // transactionTimeout must be > checkpoint interval
    .setProperty("transaction.timeout.ms", "900000")  // 15min
    .setRecordSerializer(...)
    .build();

/**
 * Implement TwoPhaseCommitSinkFunction for ANY external system.
 * Flink calls these methods at the right checkpoint phase.
 * Transaction state is checkpointed — survives restarts.
 */
public class JdbcEosSink
        extends TwoPhaseCommitSinkFunction<Revenue, Connection, Void> {

    @Override
    protected Connection beginTransaction() throws Exception {
        Connection conn = dataSource.getConnection();
        conn.setAutoCommit(false);    // start DB transaction
        return conn;
    }

    @Override
    public void invoke(Connection txn, Revenue r, Context ctx) {
        // Write into the open transaction
        try(PreparedStatement ps = txn.prepareStatement("UPSERT INTO...")){
            ps.setString(1, r.customerId);
            ps.setDouble(2, r.amount);
            ps.executeUpdate();
        }
    }

    @Override
    protected void preCommit(Connection txn) {
        // Phase 1: called when checkpoint barrier arrives
        // Flush any pending writes — do NOT commit yet
    }

    @Override
    protected void commit(Connection txn) {
        txn.commit();    // Phase 2: ALL operators confirmed
        txn.close();
    }

    @Override
    protected void abort(Connection txn) {
        txn.rollback();  // checkpoint failed → rollback
        txn.close();
    }
}

Use KAFKA STREAMS when:
  ✓ Source is exclusively Kafka topics
  ✓ Team is 1-3 engineers without dedicated ops
  ✓ State is < 10GB per application instance
  ✓ No complex temporal joins needed
  ✓ Java/Scala shop (no Python requirement)
  ✓ Microservice per-team ownership model
  ✓ < 1M msgs/sec throughput

Use FLINK when:
  ✓ Need to join Kafka with S3, JDBC, or files
  ✓ State > 10GB (RocksDB incremental checkpoints)
  ✓ Need session windows (Kafka Streams: ❌)
  ✓ Need SQL analytics on streaming data
  ✓ EOS to non-Kafka sinks (JDBC, Iceberg)
  ✓ Backfill / historical replay with correct time
  ✓ Python team / PyFlink + Table SQL API
  ✓ Multi-tenant, shared infrastructure
  ✓ Event-time correctness is non-negotiable

Use BOTH together:
  ✓ Kafka as the durable event backbone
  ✓ Kafka Streams for simple microservice transforms
  ✓ Flink for complex aggregations, ML, lake writes
  ✓ Both write results back to Kafka topics
  ← This is the production-grade architecture