Stream and batch processing

const WINDOW = 60; // tumbling windows: [0,60), [60,120), ...
const DELAY = 10; // watermark = latest event time seen - 10 s (how out of order events may be)
const LATENESS = 30; // a fired window still accepts late events until the watermark passes its end + 30 s

// [event time, arrival time] in seconds, listed in arrival order. Phones go offline; events arrive late.
const EVENTS: [number, number][] = [[5, 6], [20, 22], [59, 61], [64, 65], [55, 66], [72, 73], [50, 80], [100, 101], [30, 110], [119, 120]];

const label = (start: number) => `[${start},${start + WINDOW})`;
const windowOf = (t: number) => Math.floor(t / WINDOW) * WINDOW;

function batch(key: (e: [number, number]) => number): string {
  const counts = new Map<number, number>();
  for (const e of EVENTS) counts.set(windowOf(key(e)), (counts.get(windowOf(key(e))) ?? 0) + 1);
  return [...counts.keys()].sort((a, b) => a - b).map((s) => `${label(s)}=${counts.get(s)}`).join(" ");
}

console.log("batch, event time, all data:", batch((e) => e[0]));

const counts = new Map<number, number>();
const fired = new Set<number>();
let watermark = -Infinity;
for (const [eventTime, arrival] of EVENTS) {
  const start = windowOf(eventTime);
  if (fired.has(start) && watermark >= start + WINDOW + LATENESS) {
    console.log(`stream drops event at ${eventTime} (arrived ${arrival}): later than allowed lateness`);
    continue;
  }
  counts.set(start, (counts.get(start) ?? 0) + 1);
  if (fired.has(start)) console.log(`stream updates ${label(start)}=${counts.get(start)} for late event at ${eventTime} (arrived ${arrival})`);
  watermark = Math.max(watermark, eventTime - DELAY);
  for (const s of [...counts.keys()].sort((a, b) => a - b)) {
    if (!fired.has(s) && watermark >= s + WINDOW) {
      fired.add(s);
      console.log(`stream emits ${label(s)}=${counts.get(s)} when the watermark reaches ${watermark} (arrival ${arrival})`);
    }
  }
}
for (const s of [...counts.keys()].sort((a, b) => a - b)) {
  if (!fired.has(s)) console.log(`stream emits ${label(s)}=${counts.get(s)} at end of input`);
}

console.log("windows by arrival time instead:", batch((e) => e[1]));

import java.util.*;
import java.util.function.ToIntFunction;

public class Main {
    static final int WINDOW = 60;   // tumbling windows: [0,60), [60,120), ...
    static final int DELAY = 10;    // watermark = latest event time seen - 10 s (how out of order events may be)
    static final int LATENESS = 30; // a fired window still accepts late events until the watermark passes its end + 30 s

    // {event time, arrival time} in seconds, listed in arrival order. Phones go offline; events arrive late.
    static final int[][] EVENTS = {{5, 6}, {20, 22}, {59, 61}, {64, 65}, {55, 66}, {72, 73}, {50, 80}, {100, 101}, {30, 110}, {119, 120}};

    static String label(int start) { return "[" + start + "," + (start + WINDOW) + ")"; }

    static String batch(ToIntFunction<int[]> key) {
        TreeMap<Integer, Integer> counts = new TreeMap<>();
        for (int[] e : EVENTS) counts.merge(key.applyAsInt(e) / WINDOW * WINDOW, 1, Integer::sum);
        StringJoiner j = new StringJoiner(" ");
        counts.forEach((s, c) -> j.add(label(s) + "=" + c));
        return j.toString();
    }

    public static void main(String[] args) {
        System.out.println("batch, event time, all data: " + batch(e -> e[0]));

        TreeMap<Integer, Integer> counts = new TreeMap<>();
        Set<Integer> fired = new HashSet<>();
        long watermark = Long.MIN_VALUE;
        for (int[] e : EVENTS) {
            int eventTime = e[0], arrival = e[1], start = eventTime / WINDOW * WINDOW;
            if (fired.contains(start) && watermark >= start + WINDOW + LATENESS) {
                System.out.printf("stream drops event at %d (arrived %d): later than allowed lateness%n", eventTime, arrival);
                continue;
            }
            counts.merge(start, 1, Integer::sum);
            if (fired.contains(start)) System.out.printf("stream updates %s=%d for late event at %d (arrived %d)%n", label(start), counts.get(start), eventTime, arrival);
            watermark = Math.max(watermark, eventTime - DELAY);
            for (int s : counts.keySet()) {
                if (!fired.contains(s) && watermark >= s + WINDOW) {
                    fired.add(s);
                    System.out.printf("stream emits %s=%d when the watermark reaches %d (arrival %d)%n", label(s), counts.get(s), watermark, arrival);
                }
            }
        }
        for (int s : counts.keySet()) if (!fired.contains(s)) System.out.printf("stream emits %s=%d at end of input%n", label(s), counts.get(s));

        System.out.println("windows by arrival time instead: " + batch(e -> e[1]));
    }
}