Re: [PR] [improve][pip] PIP-484: Expose Window interface for WindowFunction processing [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #25967:
URL: https://github.com/apache/pulsar/pull/25967#discussion_r4270447097


##########
pip/pip-484.md:
##########
@@ -0,0 +1,625 @@
+# PIP-484: Expose `Window` Interface for WindowFunction Processing
+
+# Background knowledge
+
+## Pulsar Window Functions
+
+Pulsar Window Functions are a specialized form of Pulsar Function that group incoming messages into windows based on time or message count, and invoke the user function with a batch of messages each time a window fires.
+
+Window types:
+
+- **Tumbling window**: adjacent windows do not overlap; each message belongs to exactly one window.
+- **Sliding window**: adjacent windows may overlap; a message can belong to multiple windows.
+
+Time semantics:
+
+- **Processing time**: windows are driven by the clock at which messages enter the system.
+- **Event time**: windows are driven by timestamps embedded in messages, with watermarks used to track event-time progress.
+
+## Existing public API
+
+```
+pulsar-functions/api-java
+└── org.apache.pulsar.functions.api
+    ├── WindowFunction<X, T>        // user-implemented window function interface
+    └── WindowContext               // context interface for window functions
+```
+
+`WindowFunction` signature:
+
+```java
+@FunctionalInterface
+public interface WindowFunction<X, T> {
+    T process(Collection<Record<X>> input, WindowContext context) throws Exception;
+}
+```
+
+On each trigger, the user function receives a `Collection<Record<X>>` containing **all messages in the current window**.
+
+## Internal runtime pipeline
+
+```
+pulsar-functions/instance
+└── org.apache.pulsar.functions.windowing
+    ├── Window<T>                   // window view interface (internal package today; removed in Change 1)
+    ├── WindowImpl<T>               // Window implementation holding three event lists
+    ├── WindowManager<T>            // window manager; classifies events
+    ├── WindowFunctionExecutor<T,X> // executor bridging runtime and user function (T=input, X=output)
+    ├── WindowLifecycleListener<T>  // window lifecycle callbacks
+    ├── EvictionPolicy<T>           // eviction policy (decides when events expire)
+    └── TriggerPolicy<T>            // trigger policy (decides when to fire a window)
+```
+
+Today, `WindowImpl` implements an internal `org.apache.pulsar.functions.windowing.Window<T>` interface that already exposes incremental views (this PIP promotes that interface to `org.apache.pulsar.functions.api.Window` and deletes the internal copy):
+
+```java
+public interface Window<T> {
+    List<T> get();             // all events in the current window
+    List<T> getNew();          // events added since the last trigger
+    List<T> getExpired();      // events removed since the last trigger
+    Long getStartTimestamp();  // window start timestamp
+    Long getEndTimestamp();    // window end timestamp (reference time)
+}
+```
+
+# Motivation
+
+## Problem: `getNew()` / `getExpired()` data is discarded at the public API layer
+
+`WindowManager.onTrigger()` already classifies events into three categories on every window activation:
+
+| Category | Meaning |
+|----------|---------|
+| `tuples` | all events currently in the window |
+| `newTuples` | events newly added since the last trigger |
+| `expiredTuples` | events removed since the last trigger |
+
+These three lists are passed into `WindowImpl` and delivered to the executor via `WindowLifecycleListener.onActivation()`. However, `WindowFunctionExecutor.process(Window, WindowContext)` only passes `inputWindow.get()` to the user function; newly added and expired events are discarded:
+
+```java
+// WindowFunctionExecutor.java (current implementation)
+public X process(Window<Record<T>> inputWindow, WindowContext context) throws Exception {
+    // ...
+    return this.windowFunction.process(inputWindow.get(), context); // full window only; getNew()/getExpired() dropped
+}
+```
+
+## Impact
+
+Users cannot perform efficient incremental computation. Typical affected scenarios:
+
+1. **Incremental aggregation** (sliding-window statistics): on each trigger most messages in the window are unchanged; re-scanning the full collection is wasteful.
+2. **State maintenance**: when external state must track which messages entered or left the window, users must diff full collections manually — inefficient and error-prone.
+3. **Expired-event handling**: side effects such as resource release or counter decrements when messages leave the window.
+
+# Goals
+
+## In Scope
+
+- Expose the `Window<T>` interface in the public API (including `getNew()`, `getExpired()`, and timestamp methods), with explicit Javadoc for **object lifetime** (valid only during `process()`) and **list mutability** (callers must not modify returned lists; the executor snapshots the ack list so mutation cannot reach the ack path).
+- Remove the internal `org.apache.pulsar.functions.windowing.Window` type and update all instance-module references to `org.apache.pulsar.functions.api.Window`.
+- Add a new `default` overload to the existing public `WindowFunction<X, T>` interface so class-based implementations can receive the full `Window<Record<X>>` view.
+- Have `WindowFunctionExecutor` call the new `WindowFunction` overload without requiring configuration or deployment changes.
+- Deliver `onActivation` when the current window is empty but expired events remain, without invoking `process(Collection, …)`.
+- Preserve all existing behavior for current `WindowFunction` users.
+
+## Out of Scope
+
+- Incremental support for `java.util.function.Function` (bare window functions).
+- Equivalent capability for Python / Go Functions.
+- Changes to window state snapshot / checkpoint mechanisms.
+
+# High Level Design
+
+Extend the existing public `WindowFunction<X, T>` interface with a new `default` `process` overload that accepts `Window<Record<X>>` instead of only `Collection<Record<X>>`, giving class-based implementations access to:
+
+- `window.get()` — all messages in the current window
+- `window.getNew()` — messages added since the last trigger
+- `window.getExpired()` — messages removed since the last trigger
+- `window.getStartTimestamp()` / `window.getEndTimestamp()` — window time boundaries
+
+The new overload is a default method that forwards a non-empty `window.get()` to the existing abstract `process(Collection<Record<X>>, WindowContext)` method, and returns `null` when `get()` is empty:
+
+```java
+default T process(Window<Record<X>> window, WindowContext context) throws Exception {
+    List<Record<X>> current = window.get();
+    if (current.isEmpty()) {
+        return null;
+    }
+    return process(current, context);
+}
+```
+
+This keeps `WindowFunction` a valid `@FunctionalInterface`. The runtime instantiates window functions by class name, so a deployed `WindowFunction` is always a class and can override the new overload; the lambda restriction applies only to in-process/test usage.
+
+`WindowFunctionExecutor` always calls `process(Window<Record<X>>, WindowContext)`. The default overload delegates to `process(window.get(), context)` unless a class overrides the window overload.
+
+#### Empty-but-expired delivery
+
+`TimeTriggerPolicy` fires `onTrigger()` on a fixed-rate timer even after traffic stops. When the last events leave the window, `WindowManager` already drains them into `expired`, then skips `onActivation` because `events` is empty. `WindowFunctionExecutor.onExpiry` is a no-op, so those expirations never reach user code. `WindowManagerTest.testTimeBasedWindowExpiry` asserts this today.
+
+After this PIP, `WindowManager` calls `onActivation` when `events` is empty but `expired` is not. The executor still calls `process(Window, …)`:
+
+- A class that overrides `process(Window, …)` receives `get()` empty and `getExpired()` populated, so it can apply the eviction (for example decrement a running sum).
+- A class that does not override it hits the default method, which sees empty `get()` and returns `null`. The original `process(Collection, …)` is not called.
+- A bare `java.util.function.Function` is not invoked either; the executor returns `null` when `get()` is empty.
+
+### Design tradeoff
+
+Keeping `process(Collection, …)` as the single abstract method is what makes the change source- and binary-compatible. The cost is that an incremental implementation must still declare a `process(Collection, …)` body. The runtime never calls that body unless `process(Window, …)` delegates to it.
+
+That collection method should contain the full-window computation. `process(Window, …)` should call it when incremental state is missing (first activation after construction or restart). When a class overrides `process(Window, …)`, that override is the only runtime entry point.
+
+Data flow (after the change):
+
+```mermaid
+flowchart TD
+    WM["WindowManager.onTrigger()"]
+    WL["onActivation(tuples, newTuples, expiredTuples)"]
+    WE["WindowFunctionExecutor.processWindow()"]
+    WI["WindowImpl(tuples, newTuples, expiredTuples)"]
+
+    subgraph executor ["process(Window, WindowContext) dispatch"]
+        P1["WindowFunction class override<br/>→ process(window, context)"]
+        P2["WindowFunction default<br/>→ process(get()) if non-empty, else null"]
+        P3["bareWindowFunction<br/>→ apply(values) if get() non-empty"]
+    end
+
+    WM --> WL --> WE --> WI --> executor
+```
+
+# Detailed Design
+
+## Design & Implementation Details
+
+### Change 1: Move `Window<T>` to `api-java`
+
+**Current path**: `pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/Window.java`
+
+**New path**: `pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Window.java`
+
+The interface methods are unchanged in signature; the package moves to the public API module :
+
+```java
+// pulsar-functions/api-java/.../api/Window.java
+/**
+ * A per-trigger view of events in a window.
+ * <p>
+ * A {@code Window} instance is valid only for the duration of the
+ * {@link WindowFunction#process(Window, WindowContext)} call
+ * in which it is passed. Callers must not retain it across triggers.
+ * </p>
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface Window<T> {
+    /**
+     * All events in the current window for this activation.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     * <p>
+     * May be empty when this activation exists only to report expirations
+     * (see {@link #getExpired()}).
+     * </p>
+     */
+    List<T> get();
+
+    /**
+     * Events added to the window since the previous trigger.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     */
+    List<T> getNew();
+
+    /**
+     * Events removed from the window since the previous trigger.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     * <p>
+     * When the last events leave the window, {@link #get()} may be empty while
+     * this list is not.
+     * </p>
+     */
+    List<T> getExpired();
+
+    /**
+     * The window end timestamp for this activation.
+     * <p>
+     * Populated by {@code WindowFunctionExecutor} when it constructs the
+     * {@code WindowImpl} for each trigger. {@code WindowImpl} stores the value
+     * as supplied; it does not derive timestamps itself.
+     * </p>
+     * <p>
+     * For time-based windows this is the window's end boundary: the trigger
+     * policy's reference time (watermark in event-time mode, processing time in
+     * processing-time mode). For count-based windows this is the trigger policy's
+     * reference timestamp taken at trigger time, not a window boundary, and
+     * {@link #getStartTimestamp()} is {@code null}.
+     * </p>
+     */
+    Long getEndTimestamp();
+
+    /**
+     * The window start timestamp for this activation.
+     * <p>
+     * Populated by {@code WindowFunctionExecutor} when it constructs the
+     * {@code WindowImpl} for each trigger. When a time-based window length
+     * ({@code windowLengthDurationMs}) is configured, the executor sets this to
+     * {@code getEndTimestamp() - windowLengthDurationMs}. Returns {@code null}
+     * for count-based windows (no time-based window length).
+     * </p>
+     * <p>
+     * For sliding windows, start/end bound the <em>current activation's</em>
+     * window interval (window length), not the sliding step size
+     * ({@code slidingInterval*}).
+     * </p>
+     */
+    Long getStartTimestamp();
+}
+```
+
+**Migration of `org.apache.pulsar.functions.windowing.Window`**
+
+Delete the internal interface file. `WindowImpl` implements `org.apache.pulsar.functions.api.Window` directly, and every reference in `pulsar-functions/instance` is updated to import the public type:
+
+| File | Change |
+|------|--------|
+| `api-java/.../api/Window.java` | **Added** — canonical public interface (Javadoc, lifetime/mutability contracts) |
+| `windowing/Window.java` | **Removed** |
+| `windowing/WindowImpl.java` | `implements org.apache.pulsar.functions.api.Window<T>` |
+| `windowing/WindowFunctionExecutor.java` | import `org.apache.pulsar.functions.api.Window`; snapshot the ack list before `process()` |
+| `windowing/WindowFunctionExecutorTest.java` | import `org.apache.pulsar.functions.api.Window` |
+
+The old FQCN was never part of the supported public Functions API. Within this repository, only the files above referenced it; all are migrated to `org.apache.pulsar.functions.api.Window`. No deprecated shim is shipped in Apache Pulsar itself.
+
+The lists stay mutable. `get()` currently returns the same `tuples` list that `processWindow()` later iterates to ack records under `ATLEAST_ONCE`, so `window.get().clear()` would silently skip those acks. Wrapping the getters in `Collections.unmodifiableList` would turn that into `UnsupportedOperationException` for any existing `process(Collection, …)` implementation that already mutates its input, including via the default `process(Window)` overload. Instead, `processWindow()` copies `tuples` into a separate ack list **before** calling `process()`, and acks from that copy. Mutation of `get()` / `getNew()` / `getExpired()` then cannot reach the ack path or `WindowManager`'s live event store. Javadoc still tells callers not to modify the lists; that rule is advisory, not load-bearing.
+
+### Change 2: Add a `Window<Record<X>>` default overload to `WindowFunction<X, T>`
+
+**Path**: `pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/WindowFunction.java`
+
+```java
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+@FunctionalInterface
+public interface WindowFunction<X, T> {
+    /**
+     * Process all records in the triggered window.
+     *
+     * <p>This remains the single abstract method of the interface, so existing
+     * implementations continue to work without source changes. Incremental
+     * implementations should put the full-window computation here and have
+     * {@link #process(Window, WindowContext)} fall back to it when incremental
+     * state is not available.</p>
+     *
+     * <p>The runtime does not invoke this method when the current window is
+     * empty.</p>
+     *
+     * @param input   all current records in the window
+     * @param context the window function context
+     * @return the output, or {@code null} to suppress output
+     */
+    T process(Collection<Record<X>> input, WindowContext context) throws Exception;
+
+    /**
+     * Process the triggered window with access to incremental window views.
+     *
+     * <p>The default implementation preserves existing behavior: if
+     * {@link Window#get()} is empty it returns {@code null} without calling
+     * {@link #process(Collection, WindowContext)}; otherwise it delegates with
+     * {@link Window#get()}.</p>
+     *
+     * <p>Implementations can override this method to access newly added events
+     * ({@link Window#getNew()}) and expired events ({@link Window#getExpired()}).
+     * When this method is overridden, it is the runtime entry point; the
+     * collection-based method is invoked only if this override delegates to it.</p>
+     *
+     * @param window  the window view for this activation, valid only for the
+     *                duration of this method call. Callers must not retain the
+     *                {@code Window} reference or any list returned by its getters
+     *                across invocations.
+     * @param context     the window function context
+     * @return the output, or {@code null} to suppress output
+     */
+    default T process(Window<Record<X>> window, WindowContext context) throws Exception {
+        List<Record<X>> current = window.get();
+        if (current.isEmpty()) {
+            return null;
+        }
+        return process(current, context);
+    }
+}
+```
+
+Because the new method is `default`, adding it is source-compatible and binary-compatible for existing `WindowFunction` implementations.
+
+#### Example: sliding-window sum
+
+The collection method holds the full-window computation. The window overload uses `getNew()` / `getExpired()` when a running sum is already in memory, and falls back to the collection method on the first activation after construction or restart. The runtime invokes only `process(Window, …)`. The incremental update is correct when `slidingInterval ≤ windowLength`. `slidingInterval > windowLength` remains a legal config; in that case a record can expire between triggers and appear only in `getExpired()`, so a decrement-on-expiry aggregate would drift.
+
+```java
+/**
+ * Maintains the sum of integer values in the current sliding window incrementally.
+ */
+public class SlidingWindowSumFunction implements WindowFunction<Integer, Integer> {
+
+    private Long runningSum;
+
+    @Override
+    public Integer process(Collection<Record<Integer>> input, WindowContext context) {
+        int sum = 0;
+        for (Record<Integer> record : input) {
+            sum += record.getValue();
+        }
+        return sum;
+    }
+
+    @Override
+    public Integer process(Window<Record<Integer>> window, WindowContext context) throws Exception {
+        if (runningSum == null) {
+            Collection<Record<Integer>> current = window.get();
+            if (current.isEmpty()) {
+                runningSum = 0L;
+                return 0;
+            }
+            runningSum = process(current, context).longValue();
+            return runningSum.intValue();
+        }
+        for (Record<Integer> record : window.getNew()) {
+            runningSum += record.getValue();
+        }
+        for (Record<Integer> record : window.getExpired()) {
+            runningSum -= record.getValue();
+        }
+        return runningSum.intValue();
+    }
+}
+```
+
+### Change 3: Deliver empty-but-expired activations in `WindowManager`
+
+**Path**: `pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowManager.java`
+
+Today `onTrigger()` drains `expiredEvents` and then calls `onActivation` only when `events` is non-empty. If the window has drained, the expired list is dropped. `WindowFunctionExecutor.onExpiry` is a no-op, so user code never sees those events.
+
+Call `onActivation` when `events` is empty but `expired` is not. `prevWindowEvents` stays empty after the existing `clear()` when there is nothing left in the window. `triggerPolicy.reset()` still runs on every trigger.
+
+```java
+prevWindowEvents.clear();
+if (!events.isEmpty() || !expired.isEmpty()) {
+    if (!events.isEmpty()) {
+        prevWindowEvents.addAll(windowEvents);
+    }
+    windowLifecycleListener.onActivation(events, newEvents, expired,
+            evictionPolicy.getContext().getReferenceTime());
+} else {
+    log.debug("No events in the window, skipping onActivation");
+}
+```
+
+Fully empty triggers (`events` and `expired` both empty) still skip `onActivation`.
+
+### Change 4: Update `WindowFunctionExecutor`
+
+**Path**: `pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java`
+
+No new field or user-function detection path is needed. `initializeUserFunction()` continues to accept `WindowFunction` exactly as it does today. The class remains `WindowFunctionExecutor<T, X>` (T=input, X=output); this PIP does not rename its type parameters. `onExpiry` stays a no-op.
+
+Two call-site changes:
+
+1. `processWindow()` snapshots the ack list before calling `process()`, so user mutation of `window.get()` cannot skip `ATLEAST_ONCE` acks. When `tuples` is empty the ack copy is empty. This activation does not ack `getExpired()`. Under `ATLEAST_ONCE`, a record is acked only if some activation included it in a non-empty `get()`.
+
+```java
+List<Record<T>> ackTuples = new ArrayList<>(tuples);
+output = this.process(
+        new WindowImpl<>(tuples, newTuples, expiredTuples, getWindowStartTs(referenceTime), referenceTime),
+        new WindowContextImpl(context));
+// ack from ackTuples, not tuples
+```
+
+2. `process(Window<Record<T>>, WindowContext)` passes the full window view to `WindowFunction`. Bare functions return `null` when `get()` is empty:
+
+```java
+public X process(Window<Record<T>> inputWindow, WindowContext context) throws Exception {
+    if (this.bareWindowFunction != null) {
+        if (inputWindow.get().isEmpty()) {
+            return null;
+        }
+        Collection<T> values = inputWindow.get().stream()
+                .map(Record::getValue).collect(Collectors.toList());
+        return this.bareWindowFunction.apply(values);
+    } else {
+        return this.windowFunction.process(inputWindow, context);
+    }
+}
+```
+
+## Public-facing Changes
+
+### Public API
+
+#### New interface: `org.apache.pulsar.functions.api.Window<T>`
+
+Promoted from the internal package. The internal `org.apache.pulsar.functions.windowing.Window` type is removed (see Change 1); instance code imports the public interface directly. Marked `@InterfaceStability.Evolving` (same as `StateStore` / `StateStoreContext`); `WindowFunction` stays `@InterfaceStability.Stable`.
+
+| Method | Description |
+|--------|-------------|
+| `List<T> get()` | All events in the current window. Returns a mutable list that is a **snapshot** for this activation; callers must not modify it or retain it past `process()`. Mutation does not affect live window state or acks. May be empty when the activation exists only to report expirations. |
+| `List<T> getNew()` | Events added since the last trigger. Same lifetime and mutability contract as `get()`. |
+| `List<T> getExpired()` | Events removed since the last trigger. Same lifetime and mutability contract as `get()`. When the window has drained, this list may be non-empty while `get()` is empty. |
+| `Long getStartTimestamp()` | Window start time for this activation. Set by `WindowFunctionExecutor` to `getEndTimestamp() - windowLengthDurationMs` when a time-based window length is configured; `null` for count-based windows. For sliding windows, this bounds the current activation's window interval (window length), not the sliding step. |
+| `Long getEndTimestamp()` | Window end time for this activation. Set by `WindowFunctionExecutor` from the trigger policy's reference time. For time-based windows this is the window's end boundary (watermark in event-time mode, processing time in processing-time mode). For count-based windows this is the trigger policy's reference timestamp taken at trigger time, not a window boundary, and `getStartTimestamp()` is `null`. `WindowImpl` stores the value as supplied. |
+
+#### API contracts (`Window` lifetime and list mutability)
+
+On each window trigger, `WindowFunctionExecutor` builds a new `WindowImpl` with freshly collected lists, injects `getEndTimestamp()` from the trigger's reference time and `getStartTimestamp()` from `referenceTime - windowLengthDurationMs` (when applicable), and passes the view to `WindowFunction.process(Window<Record<X>>, WindowContext)`. The runtime does not retain that object after `process()` returns.
+
+| Contract | Rule |
+|----------|------|
+| **`Window` lifetime** | The `inputWindow` argument is valid **only for the duration of a single `process()` call**. Implementations must not store the `Window` reference (or any list obtained from it) in a field, cache, or closure for use on a later trigger. Each activation receives a new `Window` instance. |
+| **List mutability** | `get()`, `getNew()`, and `getExpired()` return mutable `List` instances that are **snapshots for the current activation**. Callers **must not modify** them. `processWindow()` copies `tuples` before `process()`, so mutation cannot skip `ATLEAST_ONCE` acks or update `WindowManager` live state. Javadoc on each getter states this explicitly. |
+| **Empty-but-expired delivery** | If `get()` is empty and `getExpired()` is not, `WindowManager` still calls `onActivation`. The executor calls `process(Window, …)` only. The default `process(Window, …)` returns `null` without calling `process(Collection, …)`. Bare `Function` implementations are not invoked. |
+
+These contracts mirror how `WindowFunction` already treats `Collection<Record<X>>` from `inputWindow.get()`: a per-trigger view, not durable user-owned state.
+
+#### Updated interface: `org.apache.pulsar.functions.api.WindowFunction<X, T>`
+
+The existing public interface gains a new default overload.
+
+| Method | Description |
+|--------|-------------|
+| `T process(Collection<Record<X>> input, WindowContext context)` | Existing abstract method. Existing implementations continue to target this method. Not invoked when `window.get()` is empty. |
+| `default T process(Window<Record<X>> window, WindowContext context)` | New default method. Returns `null` when `get()` is empty; otherwise delegates to `process(window.get(), context)` unless a class override provides incremental access (see **Design tradeoff**). |
+
+### Configuration
+
+No new `WindowConfig` fields or CLI options. Existing window settings (`windowLength*`, `slidingInterval*`, event-time options, etc.) apply unchanged.
+
+No submit-time validation change is required because users still implement `WindowFunction`, which is already accepted by `FunctionConfigUtils.doJavaChecks()` and `FunctionCommon.getFunctionClassParent()`.
+
+
+# Monitoring
+
+No new metrics are introduced by this change. Event classification (`get()`, `getNew()`, `getExpired()`) is unchanged.
+
+
+# Security Considerations
+
+This proposal adds a public Java API for Functions window processing only. It introduces no new REST endpoints, wire-protocol commands, authentication/authorization checks, or multi-tenancy behavior. The exposed `Window` contents are limited to records already delivered to the user function by the existing window runtime.
+
+
+# Alternatives
+
+## Alternative 1: Add a new `IncrementalWindowFunction<X, T>` interface
+
+An earlier design introduced a new public interface:
+
+```java
+@FunctionalInterface
+public interface IncrementalWindowFunction<X, T> {
+    T process(Window<Record<X>> window, WindowContext context) throws Exception;
+}
+```
+
+`WindowFunctionExecutor` would detect whether the user function implements `IncrementalWindowFunction` and dispatch to it before falling back to the existing `WindowFunction` path.
+
+Although the interface name is `IncrementalWindowFunction`, it also exposes **expired** events (via `Window#getExpired()`), since incremental window state updates may depend on both arrivals and evictions from the window.
+
+**Advantages**
+
+- Gives the incremental API a distinct type name, making user intent explicit at the class declaration.
+
+**Disadvantages**
+
+- Adds another public user-function interface for a behavior that is still semantically a window function, increasing API surface area.
+- Requires additional submit-time validation and type-inference changes.
+- If a user class implements multiple recognized interfaces (for example both `IncrementalWindowFunction` and `WindowFunction`), the runtime must define and document a dispatch precedence order. That ordering becomes an observable behavior contract and can surprise users.
+
+The selected design avoids those costs by extending the existing `WindowFunction` interface with a default overload. Existing validation and type inference continue to work because users still implement `WindowFunction`. Deployed functions are instantiated by class name, so they can override `process(Window<Record<X>>, WindowContext)`.
+
+## Alternative 2: Expose `Window` via `WindowContext`
+
+Another option is to keep the existing `WindowFunction` signature unchanged and add a getter on `WindowContext`:
+
+```java
+public interface WindowContext extends BaseContext {
+    // existing methods unchanged ...
+
+    /**
+     * The window view for the current activation.
+     * Valid only for the duration of the enclosing {@code process()} call.
+     */
+    Window<Record<?>> getWindow();
+}
+```
+
+Users would continue to implement `process(Collection<Record<X>> input, WindowContext context)` and access incremental data through the context:
+
+```java
+@Override
+public Integer process(Collection<Record<Integer>> input, WindowContext context) throws Exception {
+    Window<Record<Integer>> window = (Window) context.getWindow();
+    for (Record<Integer> record : window.getNew()) { /* ... */ }
+    return aggregate(input);
+}
+```
+
+`WindowFunctionExecutor` would attach the current `WindowImpl` to `WindowContextImpl` before calling `process()`.
+
+**Advantages**
+
+- No change to the `WindowFunction` method signature; existing lambdas and compiled user code remain fully compatible.
+- Single runtime entry point — no ambiguity between two `process` overloads.
+- Simpler executor dispatch: always call `process(inputWindow.get(), context)`; no default-method or class-loader concerns.
+
+**Disadvantages**
+
+- **Redundant inputs**: the `input` parameter is effectively `getWindow().get()`. Users must choose which source to trust, and the API must guarantee they stay in sync for the lifetime of the call.
+- **Weaker typing**: `WindowContext` cannot be generically parameterized by `X`, so `getWindow()` would return `Window<Record<?>>` (or require an unchecked cast), unlike `Window<Record<X>>` on the `process` parameter.
+- **Lower discoverability**: incremental access is hidden behind a context getter rather than expressed in the method signature.
+- **Higher lifetime risk**: developers often store or pass `Context` objects to helpers or async callbacks; attaching a per-trigger `Window` to context makes it easier to accidentally retain window state across triggers.
+
+The selected design treats the window view as **input data** and passes it explicitly via a `process(Window, WindowContext)` parameter. That keeps lifetime and mutability contracts on the parameter (where Java users expect them), avoids the `input` / `getWindow().get()` duplication, and preserves full generic typing on `Window<Record<X>>`.
+
+
+# Backward & Forward Compatibility
+
+## Existing `WindowFunction` users
+
+**Fully backward compatible for collection-based implementations.** The `WindowFunctionExecutor.initializeUserFunction()` detection path for `WindowFunction` is unchanged. Existing class implementations use the new default method, which delegates to the existing collection-based method when `get()` is non-empty. Class implementations that override both methods use the `Window<Record<X>>` overload as the runtime entry point.
+
+Empty-but-expired activations are new `onActivation` calls, but they do not enter `process(Collection, …)`. The default `process(Window, …)` returns `null`, matching today's skip (no user-function invocation, no output). Implementations that override `process(Window, …)` start seeing those activations; that is opt-in to the new overload.

Review Comment:
   **[DESIGN] Empty-window activations also reach existing WindowFunctionExecutor subclasses**
   
   The compatibility note says only classes that override `process(Window, …)` on `WindowFunction` see the new activations. That misses a second override point.
   
   `processWindow()` calls the virtual `this.process(Window, WindowContext)` on the executor ([`WindowFunctionExecutor.java:230`](https://github.com/apache/pulsar/blob/26e26b7464b198d5af5dd5f9f52842834d15242b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java#L230)). A subclass of the executor skips both new guards, the `WindowFunction` default and the bare-function branch. [`WindowFunctionExecutorTest.java:56-64`](https://github.com/apache/pulsar/blob/26e26b7464b198d5af5dd5f9f52842834d15242b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutorTest.java#L56-L64) is such a subclass. With Change 3 it gets an extra call with an empty `get()`. An override that does `get().get(0)` would throw, and one that returns a value would now publish it.
   
   Please name this case in the compatibility section. One line is enough, for example: executor subclasses now also receive empty-but-expired activations. This is the one place where the change is not purely opt-in.



##########
pip/pip-484.md:
##########
@@ -0,0 +1,625 @@
+# PIP-484: Expose `Window` Interface for WindowFunction Processing
+
+# Background knowledge
+
+## Pulsar Window Functions
+
+Pulsar Window Functions are a specialized form of Pulsar Function that group incoming messages into windows based on time or message count, and invoke the user function with a batch of messages each time a window fires.
+
+Window types:
+
+- **Tumbling window**: adjacent windows do not overlap; each message belongs to exactly one window.
+- **Sliding window**: adjacent windows may overlap; a message can belong to multiple windows.
+
+Time semantics:
+
+- **Processing time**: windows are driven by the clock at which messages enter the system.
+- **Event time**: windows are driven by timestamps embedded in messages, with watermarks used to track event-time progress.
+
+## Existing public API
+
+```
+pulsar-functions/api-java
+└── org.apache.pulsar.functions.api
+    ├── WindowFunction<X, T>        // user-implemented window function interface
+    └── WindowContext               // context interface for window functions
+```
+
+`WindowFunction` signature:
+
+```java
+@FunctionalInterface
+public interface WindowFunction<X, T> {
+    T process(Collection<Record<X>> input, WindowContext context) throws Exception;
+}
+```
+
+On each trigger, the user function receives a `Collection<Record<X>>` containing **all messages in the current window**.
+
+## Internal runtime pipeline
+
+```
+pulsar-functions/instance
+└── org.apache.pulsar.functions.windowing
+    ├── Window<T>                   // window view interface (internal package today; removed in Change 1)
+    ├── WindowImpl<T>               // Window implementation holding three event lists
+    ├── WindowManager<T>            // window manager; classifies events
+    ├── WindowFunctionExecutor<T,X> // executor bridging runtime and user function (T=input, X=output)
+    ├── WindowLifecycleListener<T>  // window lifecycle callbacks
+    ├── EvictionPolicy<T>           // eviction policy (decides when events expire)
+    └── TriggerPolicy<T>            // trigger policy (decides when to fire a window)
+```
+
+Today, `WindowImpl` implements an internal `org.apache.pulsar.functions.windowing.Window<T>` interface that already exposes incremental views (this PIP promotes that interface to `org.apache.pulsar.functions.api.Window` and deletes the internal copy):
+
+```java
+public interface Window<T> {
+    List<T> get();             // all events in the current window
+    List<T> getNew();          // events added since the last trigger
+    List<T> getExpired();      // events removed since the last trigger
+    Long getStartTimestamp();  // window start timestamp
+    Long getEndTimestamp();    // window end timestamp (reference time)
+}
+```
+
+# Motivation
+
+## Problem: `getNew()` / `getExpired()` data is discarded at the public API layer
+
+`WindowManager.onTrigger()` already classifies events into three categories on every window activation:
+
+| Category | Meaning |
+|----------|---------|
+| `tuples` | all events currently in the window |
+| `newTuples` | events newly added since the last trigger |
+| `expiredTuples` | events removed since the last trigger |
+
+These three lists are passed into `WindowImpl` and delivered to the executor via `WindowLifecycleListener.onActivation()`. However, `WindowFunctionExecutor.process(Window, WindowContext)` only passes `inputWindow.get()` to the user function; newly added and expired events are discarded:
+
+```java
+// WindowFunctionExecutor.java (current implementation)
+public X process(Window<Record<T>> inputWindow, WindowContext context) throws Exception {
+    // ...
+    return this.windowFunction.process(inputWindow.get(), context); // full window only; getNew()/getExpired() dropped
+}
+```
+
+## Impact
+
+Users cannot perform efficient incremental computation. Typical affected scenarios:
+
+1. **Incremental aggregation** (sliding-window statistics): on each trigger most messages in the window are unchanged; re-scanning the full collection is wasteful.
+2. **State maintenance**: when external state must track which messages entered or left the window, users must diff full collections manually — inefficient and error-prone.
+3. **Expired-event handling**: side effects such as resource release or counter decrements when messages leave the window.
+
+# Goals
+
+## In Scope
+
+- Expose the `Window<T>` interface in the public API (including `getNew()`, `getExpired()`, and timestamp methods), with explicit Javadoc for **object lifetime** (valid only during `process()`) and **list mutability** (callers must not modify returned lists; the executor snapshots the ack list so mutation cannot reach the ack path).
+- Remove the internal `org.apache.pulsar.functions.windowing.Window` type and update all instance-module references to `org.apache.pulsar.functions.api.Window`.
+- Add a new `default` overload to the existing public `WindowFunction<X, T>` interface so class-based implementations can receive the full `Window<Record<X>>` view.
+- Have `WindowFunctionExecutor` call the new `WindowFunction` overload without requiring configuration or deployment changes.
+- Deliver `onActivation` when the current window is empty but expired events remain, without invoking `process(Collection, …)`.
+- Preserve all existing behavior for current `WindowFunction` users.
+
+## Out of Scope
+
+- Incremental support for `java.util.function.Function` (bare window functions).
+- Equivalent capability for Python / Go Functions.
+- Changes to window state snapshot / checkpoint mechanisms.
+
+# High Level Design
+
+Extend the existing public `WindowFunction<X, T>` interface with a new `default` `process` overload that accepts `Window<Record<X>>` instead of only `Collection<Record<X>>`, giving class-based implementations access to:
+
+- `window.get()` — all messages in the current window
+- `window.getNew()` — messages added since the last trigger
+- `window.getExpired()` — messages removed since the last trigger
+- `window.getStartTimestamp()` / `window.getEndTimestamp()` — window time boundaries
+
+The new overload is a default method that forwards a non-empty `window.get()` to the existing abstract `process(Collection<Record<X>>, WindowContext)` method, and returns `null` when `get()` is empty:
+
+```java
+default T process(Window<Record<X>> window, WindowContext context) throws Exception {
+    List<Record<X>> current = window.get();
+    if (current.isEmpty()) {
+        return null;
+    }
+    return process(current, context);
+}
+```
+
+This keeps `WindowFunction` a valid `@FunctionalInterface`. The runtime instantiates window functions by class name, so a deployed `WindowFunction` is always a class and can override the new overload; the lambda restriction applies only to in-process/test usage.
+
+`WindowFunctionExecutor` always calls `process(Window<Record<X>>, WindowContext)`. The default overload delegates to `process(window.get(), context)` unless a class overrides the window overload.
+
+#### Empty-but-expired delivery
+
+`TimeTriggerPolicy` fires `onTrigger()` on a fixed-rate timer even after traffic stops. When the last events leave the window, `WindowManager` already drains them into `expired`, then skips `onActivation` because `events` is empty. `WindowFunctionExecutor.onExpiry` is a no-op, so those expirations never reach user code. `WindowManagerTest.testTimeBasedWindowExpiry` asserts this today.
+
+After this PIP, `WindowManager` calls `onActivation` when `events` is empty but `expired` is not. The executor still calls `process(Window, …)`:
+
+- A class that overrides `process(Window, …)` receives `get()` empty and `getExpired()` populated, so it can apply the eviction (for example decrement a running sum).
+- A class that does not override it hits the default method, which sees empty `get()` and returns `null`. The original `process(Collection, …)` is not called.
+- A bare `java.util.function.Function` is not invoked either; the executor returns `null` when `get()` is empty.
+
+### Design tradeoff
+
+Keeping `process(Collection, …)` as the single abstract method is what makes the change source- and binary-compatible. The cost is that an incremental implementation must still declare a `process(Collection, …)` body. The runtime never calls that body unless `process(Window, …)` delegates to it.
+
+That collection method should contain the full-window computation. `process(Window, …)` should call it when incremental state is missing (first activation after construction or restart). When a class overrides `process(Window, …)`, that override is the only runtime entry point.
+
+Data flow (after the change):
+
+```mermaid
+flowchart TD
+    WM["WindowManager.onTrigger()"]
+    WL["onActivation(tuples, newTuples, expiredTuples)"]
+    WE["WindowFunctionExecutor.processWindow()"]
+    WI["WindowImpl(tuples, newTuples, expiredTuples)"]
+
+    subgraph executor ["process(Window, WindowContext) dispatch"]
+        P1["WindowFunction class override<br/>→ process(window, context)"]
+        P2["WindowFunction default<br/>→ process(get()) if non-empty, else null"]
+        P3["bareWindowFunction<br/>→ apply(values) if get() non-empty"]
+    end
+
+    WM --> WL --> WE --> WI --> executor
+```
+
+# Detailed Design
+
+## Design & Implementation Details
+
+### Change 1: Move `Window<T>` to `api-java`
+
+**Current path**: `pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/Window.java`
+
+**New path**: `pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Window.java`
+
+The interface methods are unchanged in signature; the package moves to the public API module :
+
+```java
+// pulsar-functions/api-java/.../api/Window.java
+/**
+ * A per-trigger view of events in a window.
+ * <p>
+ * A {@code Window} instance is valid only for the duration of the
+ * {@link WindowFunction#process(Window, WindowContext)} call
+ * in which it is passed. Callers must not retain it across triggers.
+ * </p>
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface Window<T> {
+    /**
+     * All events in the current window for this activation.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     * <p>
+     * May be empty when this activation exists only to report expirations
+     * (see {@link #getExpired()}).
+     * </p>
+     */
+    List<T> get();
+
+    /**
+     * Events added to the window since the previous trigger.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     */
+    List<T> getNew();
+
+    /**
+     * Events removed from the window since the previous trigger.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     * <p>
+     * When the last events leave the window, {@link #get()} may be empty while
+     * this list is not.
+     * </p>
+     */
+    List<T> getExpired();
+
+    /**
+     * The window end timestamp for this activation.
+     * <p>
+     * Populated by {@code WindowFunctionExecutor} when it constructs the
+     * {@code WindowImpl} for each trigger. {@code WindowImpl} stores the value
+     * as supplied; it does not derive timestamps itself.
+     * </p>
+     * <p>
+     * For time-based windows this is the window's end boundary: the trigger
+     * policy's reference time (watermark in event-time mode, processing time in
+     * processing-time mode). For count-based windows this is the trigger policy's
+     * reference timestamp taken at trigger time, not a window boundary, and
+     * {@link #getStartTimestamp()} is {@code null}.
+     * </p>
+     */
+    Long getEndTimestamp();
+
+    /**
+     * The window start timestamp for this activation.
+     * <p>
+     * Populated by {@code WindowFunctionExecutor} when it constructs the
+     * {@code WindowImpl} for each trigger. When a time-based window length
+     * ({@code windowLengthDurationMs}) is configured, the executor sets this to
+     * {@code getEndTimestamp() - windowLengthDurationMs}. Returns {@code null}
+     * for count-based windows (no time-based window length).
+     * </p>
+     * <p>
+     * For sliding windows, start/end bound the <em>current activation's</em>
+     * window interval (window length), not the sliding step size
+     * ({@code slidingInterval*}).
+     * </p>
+     */
+    Long getStartTimestamp();
+}
+```
+
+**Migration of `org.apache.pulsar.functions.windowing.Window`**
+
+Delete the internal interface file. `WindowImpl` implements `org.apache.pulsar.functions.api.Window` directly, and every reference in `pulsar-functions/instance` is updated to import the public type:
+
+| File | Change |
+|------|--------|
+| `api-java/.../api/Window.java` | **Added** — canonical public interface (Javadoc, lifetime/mutability contracts) |
+| `windowing/Window.java` | **Removed** |
+| `windowing/WindowImpl.java` | `implements org.apache.pulsar.functions.api.Window<T>` |
+| `windowing/WindowFunctionExecutor.java` | import `org.apache.pulsar.functions.api.Window`; snapshot the ack list before `process()` |
+| `windowing/WindowFunctionExecutorTest.java` | import `org.apache.pulsar.functions.api.Window` |
+
+The old FQCN was never part of the supported public Functions API. Within this repository, only the files above referenced it; all are migrated to `org.apache.pulsar.functions.api.Window`. No deprecated shim is shipped in Apache Pulsar itself.
+
+The lists stay mutable. `get()` currently returns the same `tuples` list that `processWindow()` later iterates to ack records under `ATLEAST_ONCE`, so `window.get().clear()` would silently skip those acks. Wrapping the getters in `Collections.unmodifiableList` would turn that into `UnsupportedOperationException` for any existing `process(Collection, …)` implementation that already mutates its input, including via the default `process(Window)` overload. Instead, `processWindow()` copies `tuples` into a separate ack list **before** calling `process()`, and acks from that copy. Mutation of `get()` / `getNew()` / `getExpired()` then cannot reach the ack path or `WindowManager`'s live event store. Javadoc still tells callers not to modify the lists; that rule is advisory, not load-bearing.
+
+### Change 2: Add a `Window<Record<X>>` default overload to `WindowFunction<X, T>`
+
+**Path**: `pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/WindowFunction.java`
+
+```java
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+@FunctionalInterface
+public interface WindowFunction<X, T> {
+    /**
+     * Process all records in the triggered window.
+     *
+     * <p>This remains the single abstract method of the interface, so existing
+     * implementations continue to work without source changes. Incremental
+     * implementations should put the full-window computation here and have
+     * {@link #process(Window, WindowContext)} fall back to it when incremental
+     * state is not available.</p>
+     *
+     * <p>The runtime does not invoke this method when the current window is
+     * empty.</p>
+     *
+     * @param input   all current records in the window
+     * @param context the window function context
+     * @return the output, or {@code null} to suppress output
+     */
+    T process(Collection<Record<X>> input, WindowContext context) throws Exception;
+
+    /**
+     * Process the triggered window with access to incremental window views.
+     *
+     * <p>The default implementation preserves existing behavior: if
+     * {@link Window#get()} is empty it returns {@code null} without calling
+     * {@link #process(Collection, WindowContext)}; otherwise it delegates with
+     * {@link Window#get()}.</p>
+     *
+     * <p>Implementations can override this method to access newly added events
+     * ({@link Window#getNew()}) and expired events ({@link Window#getExpired()}).
+     * When this method is overridden, it is the runtime entry point; the
+     * collection-based method is invoked only if this override delegates to it.</p>
+     *
+     * @param window  the window view for this activation, valid only for the
+     *                duration of this method call. Callers must not retain the
+     *                {@code Window} reference or any list returned by its getters
+     *                across invocations.
+     * @param context     the window function context
+     * @return the output, or {@code null} to suppress output
+     */
+    default T process(Window<Record<X>> window, WindowContext context) throws Exception {
+        List<Record<X>> current = window.get();
+        if (current.isEmpty()) {
+            return null;
+        }
+        return process(current, context);
+    }
+}
+```
+
+Because the new method is `default`, adding it is source-compatible and binary-compatible for existing `WindowFunction` implementations.
+
+#### Example: sliding-window sum
+
+The collection method holds the full-window computation. The window overload uses `getNew()` / `getExpired()` when a running sum is already in memory, and falls back to the collection method on the first activation after construction or restart. The runtime invokes only `process(Window, …)`. The incremental update is correct when `slidingInterval ≤ windowLength`. `slidingInterval > windowLength` remains a legal config; in that case a record can expire between triggers and appear only in `getExpired()`, so a decrement-on-expiry aggregate would drift.

Review Comment:
   **[DESIGN] The slidingInterval <= windowLength condition is stated as sufficient, but it is not for event time**
   
   The new sentence says the incremental update is correct when `slidingInterval ≤ windowLength`. That is a good caveat, but it reads as a guarantee and it is too strong.
   
   In event-time mode [`WatermarkTimeEvictionPolicy.java:74`](https://github.com/apache/pulsar/blob/26e26b7464b198d5af5dd5f9f52842834d15242b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/evictors/WatermarkTimeEvictionPolicy.java#L74) returns `STOP` for a record that is far enough in the future, and [`WindowManager.java:208`](https://github.com/apache/pulsar/blob/26e26b7464b198d5af5dd5f9f52842834d15242b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowManager.java#L208) stops the scan on `STOP`. A late record that sits behind such a record can expire without ever being reported in `get()` or `getNew()`. It is accepted because [`WaterMarkEventGenerator.java:90`](https://github.com/apache/pulsar/blob/26e26b7464b198d5af5dd5f9f52842834d15242b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WaterMarkEventGenerator.java#L90) compares against the last emitted watermark. Then `getExpired()` holds 
 a record the function never added, and the running sum drifts.
   
   Suggestion: say the example is exact only for processing-time windows with `slidingInterval ≤ windowLength`, and that event-time windows with late records need the function to track what it has seen. Or reword to say the condition is necessary, not sufficient. This also fits the author's plan to forbid slide > window: the PIP still calls it legal, so please keep the two statements consistent.



##########
pip/pip-484.md:
##########
@@ -0,0 +1,625 @@
+# PIP-484: Expose `Window` Interface for WindowFunction Processing
+
+# Background knowledge
+
+## Pulsar Window Functions
+
+Pulsar Window Functions are a specialized form of Pulsar Function that group incoming messages into windows based on time or message count, and invoke the user function with a batch of messages each time a window fires.
+
+Window types:
+
+- **Tumbling window**: adjacent windows do not overlap; each message belongs to exactly one window.
+- **Sliding window**: adjacent windows may overlap; a message can belong to multiple windows.
+
+Time semantics:
+
+- **Processing time**: windows are driven by the clock at which messages enter the system.
+- **Event time**: windows are driven by timestamps embedded in messages, with watermarks used to track event-time progress.
+
+## Existing public API
+
+```
+pulsar-functions/api-java
+└── org.apache.pulsar.functions.api
+    ├── WindowFunction<X, T>        // user-implemented window function interface
+    └── WindowContext               // context interface for window functions
+```
+
+`WindowFunction` signature:
+
+```java
+@FunctionalInterface
+public interface WindowFunction<X, T> {
+    T process(Collection<Record<X>> input, WindowContext context) throws Exception;
+}
+```
+
+On each trigger, the user function receives a `Collection<Record<X>>` containing **all messages in the current window**.
+
+## Internal runtime pipeline
+
+```
+pulsar-functions/instance
+└── org.apache.pulsar.functions.windowing
+    ├── Window<T>                   // window view interface (internal package today; removed in Change 1)
+    ├── WindowImpl<T>               // Window implementation holding three event lists
+    ├── WindowManager<T>            // window manager; classifies events
+    ├── WindowFunctionExecutor<T,X> // executor bridging runtime and user function (T=input, X=output)
+    ├── WindowLifecycleListener<T>  // window lifecycle callbacks
+    ├── EvictionPolicy<T>           // eviction policy (decides when events expire)
+    └── TriggerPolicy<T>            // trigger policy (decides when to fire a window)
+```
+
+Today, `WindowImpl` implements an internal `org.apache.pulsar.functions.windowing.Window<T>` interface that already exposes incremental views (this PIP promotes that interface to `org.apache.pulsar.functions.api.Window` and deletes the internal copy):
+
+```java
+public interface Window<T> {
+    List<T> get();             // all events in the current window
+    List<T> getNew();          // events added since the last trigger
+    List<T> getExpired();      // events removed since the last trigger
+    Long getStartTimestamp();  // window start timestamp
+    Long getEndTimestamp();    // window end timestamp (reference time)
+}
+```
+
+# Motivation
+
+## Problem: `getNew()` / `getExpired()` data is discarded at the public API layer
+
+`WindowManager.onTrigger()` already classifies events into three categories on every window activation:
+
+| Category | Meaning |
+|----------|---------|
+| `tuples` | all events currently in the window |
+| `newTuples` | events newly added since the last trigger |
+| `expiredTuples` | events removed since the last trigger |
+
+These three lists are passed into `WindowImpl` and delivered to the executor via `WindowLifecycleListener.onActivation()`. However, `WindowFunctionExecutor.process(Window, WindowContext)` only passes `inputWindow.get()` to the user function; newly added and expired events are discarded:
+
+```java
+// WindowFunctionExecutor.java (current implementation)
+public X process(Window<Record<T>> inputWindow, WindowContext context) throws Exception {
+    // ...
+    return this.windowFunction.process(inputWindow.get(), context); // full window only; getNew()/getExpired() dropped
+}
+```
+
+## Impact
+
+Users cannot perform efficient incremental computation. Typical affected scenarios:
+
+1. **Incremental aggregation** (sliding-window statistics): on each trigger most messages in the window are unchanged; re-scanning the full collection is wasteful.
+2. **State maintenance**: when external state must track which messages entered or left the window, users must diff full collections manually — inefficient and error-prone.
+3. **Expired-event handling**: side effects such as resource release or counter decrements when messages leave the window.
+
+# Goals
+
+## In Scope
+
+- Expose the `Window<T>` interface in the public API (including `getNew()`, `getExpired()`, and timestamp methods), with explicit Javadoc for **object lifetime** (valid only during `process()`) and **list mutability** (callers must not modify returned lists; the executor snapshots the ack list so mutation cannot reach the ack path).
+- Remove the internal `org.apache.pulsar.functions.windowing.Window` type and update all instance-module references to `org.apache.pulsar.functions.api.Window`.
+- Add a new `default` overload to the existing public `WindowFunction<X, T>` interface so class-based implementations can receive the full `Window<Record<X>>` view.
+- Have `WindowFunctionExecutor` call the new `WindowFunction` overload without requiring configuration or deployment changes.
+- Deliver `onActivation` when the current window is empty but expired events remain, without invoking `process(Collection, …)`.
+- Preserve all existing behavior for current `WindowFunction` users.
+
+## Out of Scope
+
+- Incremental support for `java.util.function.Function` (bare window functions).
+- Equivalent capability for Python / Go Functions.
+- Changes to window state snapshot / checkpoint mechanisms.
+
+# High Level Design
+
+Extend the existing public `WindowFunction<X, T>` interface with a new `default` `process` overload that accepts `Window<Record<X>>` instead of only `Collection<Record<X>>`, giving class-based implementations access to:
+
+- `window.get()` — all messages in the current window
+- `window.getNew()` — messages added since the last trigger
+- `window.getExpired()` — messages removed since the last trigger
+- `window.getStartTimestamp()` / `window.getEndTimestamp()` — window time boundaries
+
+The new overload is a default method that forwards a non-empty `window.get()` to the existing abstract `process(Collection<Record<X>>, WindowContext)` method, and returns `null` when `get()` is empty:
+
+```java
+default T process(Window<Record<X>> window, WindowContext context) throws Exception {
+    List<Record<X>> current = window.get();
+    if (current.isEmpty()) {
+        return null;
+    }
+    return process(current, context);
+}
+```
+
+This keeps `WindowFunction` a valid `@FunctionalInterface`. The runtime instantiates window functions by class name, so a deployed `WindowFunction` is always a class and can override the new overload; the lambda restriction applies only to in-process/test usage.
+
+`WindowFunctionExecutor` always calls `process(Window<Record<X>>, WindowContext)`. The default overload delegates to `process(window.get(), context)` unless a class overrides the window overload.
+
+#### Empty-but-expired delivery
+
+`TimeTriggerPolicy` fires `onTrigger()` on a fixed-rate timer even after traffic stops. When the last events leave the window, `WindowManager` already drains them into `expired`, then skips `onActivation` because `events` is empty. `WindowFunctionExecutor.onExpiry` is a no-op, so those expirations never reach user code. `WindowManagerTest.testTimeBasedWindowExpiry` asserts this today.
+
+After this PIP, `WindowManager` calls `onActivation` when `events` is empty but `expired` is not. The executor still calls `process(Window, …)`:
+
+- A class that overrides `process(Window, …)` receives `get()` empty and `getExpired()` populated, so it can apply the eviction (for example decrement a running sum).
+- A class that does not override it hits the default method, which sees empty `get()` and returns `null`. The original `process(Collection, …)` is not called.
+- A bare `java.util.function.Function` is not invoked either; the executor returns `null` when `get()` is empty.
+
+### Design tradeoff
+
+Keeping `process(Collection, …)` as the single abstract method is what makes the change source- and binary-compatible. The cost is that an incremental implementation must still declare a `process(Collection, …)` body. The runtime never calls that body unless `process(Window, …)` delegates to it.
+
+That collection method should contain the full-window computation. `process(Window, …)` should call it when incremental state is missing (first activation after construction or restart). When a class overrides `process(Window, …)`, that override is the only runtime entry point.
+
+Data flow (after the change):
+
+```mermaid
+flowchart TD
+    WM["WindowManager.onTrigger()"]
+    WL["onActivation(tuples, newTuples, expiredTuples)"]
+    WE["WindowFunctionExecutor.processWindow()"]
+    WI["WindowImpl(tuples, newTuples, expiredTuples)"]
+
+    subgraph executor ["process(Window, WindowContext) dispatch"]
+        P1["WindowFunction class override<br/>→ process(window, context)"]
+        P2["WindowFunction default<br/>→ process(get()) if non-empty, else null"]
+        P3["bareWindowFunction<br/>→ apply(values) if get() non-empty"]
+    end
+
+    WM --> WL --> WE --> WI --> executor
+```
+
+# Detailed Design
+
+## Design & Implementation Details
+
+### Change 1: Move `Window<T>` to `api-java`
+
+**Current path**: `pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/Window.java`
+
+**New path**: `pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Window.java`
+
+The interface methods are unchanged in signature; the package moves to the public API module :
+
+```java
+// pulsar-functions/api-java/.../api/Window.java
+/**
+ * A per-trigger view of events in a window.
+ * <p>
+ * A {@code Window} instance is valid only for the duration of the
+ * {@link WindowFunction#process(Window, WindowContext)} call
+ * in which it is passed. Callers must not retain it across triggers.
+ * </p>
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface Window<T> {
+    /**
+     * All events in the current window for this activation.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     * <p>
+     * May be empty when this activation exists only to report expirations
+     * (see {@link #getExpired()}).
+     * </p>
+     */
+    List<T> get();
+
+    /**
+     * Events added to the window since the previous trigger.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     */
+    List<T> getNew();
+
+    /**
+     * Events removed from the window since the previous trigger.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     * <p>
+     * When the last events leave the window, {@link #get()} may be empty while
+     * this list is not.
+     * </p>
+     */
+    List<T> getExpired();
+
+    /**
+     * The window end timestamp for this activation.
+     * <p>
+     * Populated by {@code WindowFunctionExecutor} when it constructs the
+     * {@code WindowImpl} for each trigger. {@code WindowImpl} stores the value
+     * as supplied; it does not derive timestamps itself.
+     * </p>
+     * <p>
+     * For time-based windows this is the window's end boundary: the trigger
+     * policy's reference time (watermark in event-time mode, processing time in
+     * processing-time mode). For count-based windows this is the trigger policy's
+     * reference timestamp taken at trigger time, not a window boundary, and
+     * {@link #getStartTimestamp()} is {@code null}.
+     * </p>
+     */
+    Long getEndTimestamp();
+
+    /**
+     * The window start timestamp for this activation.
+     * <p>
+     * Populated by {@code WindowFunctionExecutor} when it constructs the
+     * {@code WindowImpl} for each trigger. When a time-based window length
+     * ({@code windowLengthDurationMs}) is configured, the executor sets this to
+     * {@code getEndTimestamp() - windowLengthDurationMs}. Returns {@code null}
+     * for count-based windows (no time-based window length).
+     * </p>
+     * <p>
+     * For sliding windows, start/end bound the <em>current activation's</em>
+     * window interval (window length), not the sliding step size
+     * ({@code slidingInterval*}).
+     * </p>
+     */
+    Long getStartTimestamp();
+}
+```
+
+**Migration of `org.apache.pulsar.functions.windowing.Window`**
+
+Delete the internal interface file. `WindowImpl` implements `org.apache.pulsar.functions.api.Window` directly, and every reference in `pulsar-functions/instance` is updated to import the public type:
+
+| File | Change |
+|------|--------|
+| `api-java/.../api/Window.java` | **Added** — canonical public interface (Javadoc, lifetime/mutability contracts) |
+| `windowing/Window.java` | **Removed** |
+| `windowing/WindowImpl.java` | `implements org.apache.pulsar.functions.api.Window<T>` |
+| `windowing/WindowFunctionExecutor.java` | import `org.apache.pulsar.functions.api.Window`; snapshot the ack list before `process()` |
+| `windowing/WindowFunctionExecutorTest.java` | import `org.apache.pulsar.functions.api.Window` |
+
+The old FQCN was never part of the supported public Functions API. Within this repository, only the files above referenced it; all are migrated to `org.apache.pulsar.functions.api.Window`. No deprecated shim is shipped in Apache Pulsar itself.
+
+The lists stay mutable. `get()` currently returns the same `tuples` list that `processWindow()` later iterates to ack records under `ATLEAST_ONCE`, so `window.get().clear()` would silently skip those acks. Wrapping the getters in `Collections.unmodifiableList` would turn that into `UnsupportedOperationException` for any existing `process(Collection, …)` implementation that already mutates its input, including via the default `process(Window)` overload. Instead, `processWindow()` copies `tuples` into a separate ack list **before** calling `process()`, and acks from that copy. Mutation of `get()` / `getNew()` / `getExpired()` then cannot reach the ack path or `WindowManager`'s live event store. Javadoc still tells callers not to modify the lists; that rule is advisory, not load-bearing.
+
+### Change 2: Add a `Window<Record<X>>` default overload to `WindowFunction<X, T>`
+
+**Path**: `pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/WindowFunction.java`
+
+```java
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+@FunctionalInterface
+public interface WindowFunction<X, T> {
+    /**
+     * Process all records in the triggered window.
+     *
+     * <p>This remains the single abstract method of the interface, so existing
+     * implementations continue to work without source changes. Incremental
+     * implementations should put the full-window computation here and have
+     * {@link #process(Window, WindowContext)} fall back to it when incremental
+     * state is not available.</p>
+     *
+     * <p>The runtime does not invoke this method when the current window is
+     * empty.</p>
+     *
+     * @param input   all current records in the window
+     * @param context the window function context
+     * @return the output, or {@code null} to suppress output
+     */
+    T process(Collection<Record<X>> input, WindowContext context) throws Exception;
+
+    /**
+     * Process the triggered window with access to incremental window views.
+     *
+     * <p>The default implementation preserves existing behavior: if
+     * {@link Window#get()} is empty it returns {@code null} without calling
+     * {@link #process(Collection, WindowContext)}; otherwise it delegates with
+     * {@link Window#get()}.</p>
+     *
+     * <p>Implementations can override this method to access newly added events
+     * ({@link Window#getNew()}) and expired events ({@link Window#getExpired()}).
+     * When this method is overridden, it is the runtime entry point; the
+     * collection-based method is invoked only if this override delegates to it.</p>
+     *
+     * @param window  the window view for this activation, valid only for the
+     *                duration of this method call. Callers must not retain the
+     *                {@code Window} reference or any list returned by its getters
+     *                across invocations.
+     * @param context     the window function context
+     * @return the output, or {@code null} to suppress output
+     */
+    default T process(Window<Record<X>> window, WindowContext context) throws Exception {
+        List<Record<X>> current = window.get();
+        if (current.isEmpty()) {
+            return null;
+        }
+        return process(current, context);
+    }
+}
+```
+
+Because the new method is `default`, adding it is source-compatible and binary-compatible for existing `WindowFunction` implementations.
+
+#### Example: sliding-window sum
+
+The collection method holds the full-window computation. The window overload uses `getNew()` / `getExpired()` when a running sum is already in memory, and falls back to the collection method on the first activation after construction or restart. The runtime invokes only `process(Window, …)`. The incremental update is correct when `slidingInterval ≤ windowLength`. `slidingInterval > windowLength` remains a legal config; in that case a record can expire between triggers and appear only in `getExpired()`, so a decrement-on-expiry aggregate would drift.
+
+```java
+/**
+ * Maintains the sum of integer values in the current sliding window incrementally.
+ */
+public class SlidingWindowSumFunction implements WindowFunction<Integer, Integer> {
+
+    private Long runningSum;
+
+    @Override
+    public Integer process(Collection<Record<Integer>> input, WindowContext context) {
+        int sum = 0;
+        for (Record<Integer> record : input) {
+            sum += record.getValue();
+        }
+        return sum;
+    }
+
+    @Override
+    public Integer process(Window<Record<Integer>> window, WindowContext context) throws Exception {
+        if (runningSum == null) {
+            Collection<Record<Integer>> current = window.get();
+            if (current.isEmpty()) {
+                runningSum = 0L;
+                return 0;
+            }
+            runningSum = process(current, context).longValue();
+            return runningSum.intValue();
+        }
+        for (Record<Integer> record : window.getNew()) {
+            runningSum += record.getValue();
+        }
+        for (Record<Integer> record : window.getExpired()) {
+            runningSum -= record.getValue();
+        }
+        return runningSum.intValue();
+    }
+}
+```
+
+### Change 3: Deliver empty-but-expired activations in `WindowManager`
+
+**Path**: `pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowManager.java`
+
+Today `onTrigger()` drains `expiredEvents` and then calls `onActivation` only when `events` is non-empty. If the window has drained, the expired list is dropped. `WindowFunctionExecutor.onExpiry` is a no-op, so user code never sees those events.
+
+Call `onActivation` when `events` is empty but `expired` is not. `prevWindowEvents` stays empty after the existing `clear()` when there is nothing left in the window. `triggerPolicy.reset()` still runs on every trigger.
+
+```java
+prevWindowEvents.clear();
+if (!events.isEmpty() || !expired.isEmpty()) {

Review Comment:
   **[DESIGN] State that onTrigger() keeps returning !events.isEmpty()**
   
   The Change 3 snippet does not show the return value of `onTrigger()`. It matters. [`WatermarkTimeTriggerPolicy.java:90`](https://github.com/apache/pulsar/blob/26e26b7464b198d5af5dd5f9f52842834d15242b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/triggers/WatermarkTimeTriggerPolicy.java#L90) advances by one sliding interval on `true`. On `false` it searches for the next aligned window.
   
   Today the method returns `!events.isEmpty()` ([`WindowManager.java:156`](https://github.com/apache/pulsar/blob/26e26b7464b198d5af5dd5f9f52842834d15242b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowManager.java#L156)). If an implementer reads the new condition and also widens the return value to `!events.isEmpty() || !expired.isEmpty()`, event-time progression and checkpoint state change.
   
   Please add one sentence saying the return value is unchanged. Please also add an event-time test to the intended test list: an expiration-only activation, the return value, and the next window position.



##########
pip/pip-484.md:
##########
@@ -0,0 +1,625 @@
+# PIP-484: Expose `Window` Interface for WindowFunction Processing
+
+# Background knowledge
+
+## Pulsar Window Functions
+
+Pulsar Window Functions are a specialized form of Pulsar Function that group incoming messages into windows based on time or message count, and invoke the user function with a batch of messages each time a window fires.
+
+Window types:
+
+- **Tumbling window**: adjacent windows do not overlap; each message belongs to exactly one window.
+- **Sliding window**: adjacent windows may overlap; a message can belong to multiple windows.
+
+Time semantics:
+
+- **Processing time**: windows are driven by the clock at which messages enter the system.
+- **Event time**: windows are driven by timestamps embedded in messages, with watermarks used to track event-time progress.
+
+## Existing public API
+
+```
+pulsar-functions/api-java
+└── org.apache.pulsar.functions.api
+    ├── WindowFunction<X, T>        // user-implemented window function interface
+    └── WindowContext               // context interface for window functions
+```
+
+`WindowFunction` signature:
+
+```java
+@FunctionalInterface
+public interface WindowFunction<X, T> {
+    T process(Collection<Record<X>> input, WindowContext context) throws Exception;
+}
+```
+
+On each trigger, the user function receives a `Collection<Record<X>>` containing **all messages in the current window**.
+
+## Internal runtime pipeline
+
+```
+pulsar-functions/instance
+└── org.apache.pulsar.functions.windowing
+    ├── Window<T>                   // window view interface (internal package today; removed in Change 1)
+    ├── WindowImpl<T>               // Window implementation holding three event lists
+    ├── WindowManager<T>            // window manager; classifies events
+    ├── WindowFunctionExecutor<T,X> // executor bridging runtime and user function (T=input, X=output)
+    ├── WindowLifecycleListener<T>  // window lifecycle callbacks
+    ├── EvictionPolicy<T>           // eviction policy (decides when events expire)
+    └── TriggerPolicy<T>            // trigger policy (decides when to fire a window)
+```
+
+Today, `WindowImpl` implements an internal `org.apache.pulsar.functions.windowing.Window<T>` interface that already exposes incremental views (this PIP promotes that interface to `org.apache.pulsar.functions.api.Window` and deletes the internal copy):
+
+```java
+public interface Window<T> {
+    List<T> get();             // all events in the current window
+    List<T> getNew();          // events added since the last trigger
+    List<T> getExpired();      // events removed since the last trigger
+    Long getStartTimestamp();  // window start timestamp
+    Long getEndTimestamp();    // window end timestamp (reference time)
+}
+```
+
+# Motivation
+
+## Problem: `getNew()` / `getExpired()` data is discarded at the public API layer
+
+`WindowManager.onTrigger()` already classifies events into three categories on every window activation:
+
+| Category | Meaning |
+|----------|---------|
+| `tuples` | all events currently in the window |
+| `newTuples` | events newly added since the last trigger |
+| `expiredTuples` | events removed since the last trigger |
+
+These three lists are passed into `WindowImpl` and delivered to the executor via `WindowLifecycleListener.onActivation()`. However, `WindowFunctionExecutor.process(Window, WindowContext)` only passes `inputWindow.get()` to the user function; newly added and expired events are discarded:
+
+```java
+// WindowFunctionExecutor.java (current implementation)
+public X process(Window<Record<T>> inputWindow, WindowContext context) throws Exception {
+    // ...
+    return this.windowFunction.process(inputWindow.get(), context); // full window only; getNew()/getExpired() dropped
+}
+```
+
+## Impact
+
+Users cannot perform efficient incremental computation. Typical affected scenarios:
+
+1. **Incremental aggregation** (sliding-window statistics): on each trigger most messages in the window are unchanged; re-scanning the full collection is wasteful.
+2. **State maintenance**: when external state must track which messages entered or left the window, users must diff full collections manually — inefficient and error-prone.
+3. **Expired-event handling**: side effects such as resource release or counter decrements when messages leave the window.
+
+# Goals
+
+## In Scope
+
+- Expose the `Window<T>` interface in the public API (including `getNew()`, `getExpired()`, and timestamp methods), with explicit Javadoc for **object lifetime** (valid only during `process()`) and **list mutability** (callers must not modify returned lists; the executor snapshots the ack list so mutation cannot reach the ack path).
+- Remove the internal `org.apache.pulsar.functions.windowing.Window` type and update all instance-module references to `org.apache.pulsar.functions.api.Window`.
+- Add a new `default` overload to the existing public `WindowFunction<X, T>` interface so class-based implementations can receive the full `Window<Record<X>>` view.
+- Have `WindowFunctionExecutor` call the new `WindowFunction` overload without requiring configuration or deployment changes.
+- Deliver `onActivation` when the current window is empty but expired events remain, without invoking `process(Collection, …)`.
+- Preserve all existing behavior for current `WindowFunction` users.
+
+## Out of Scope
+
+- Incremental support for `java.util.function.Function` (bare window functions).
+- Equivalent capability for Python / Go Functions.
+- Changes to window state snapshot / checkpoint mechanisms.
+
+# High Level Design
+
+Extend the existing public `WindowFunction<X, T>` interface with a new `default` `process` overload that accepts `Window<Record<X>>` instead of only `Collection<Record<X>>`, giving class-based implementations access to:
+
+- `window.get()` — all messages in the current window
+- `window.getNew()` — messages added since the last trigger
+- `window.getExpired()` — messages removed since the last trigger
+- `window.getStartTimestamp()` / `window.getEndTimestamp()` — window time boundaries
+
+The new overload is a default method that forwards a non-empty `window.get()` to the existing abstract `process(Collection<Record<X>>, WindowContext)` method, and returns `null` when `get()` is empty:
+
+```java
+default T process(Window<Record<X>> window, WindowContext context) throws Exception {
+    List<Record<X>> current = window.get();
+    if (current.isEmpty()) {
+        return null;
+    }
+    return process(current, context);
+}
+```
+
+This keeps `WindowFunction` a valid `@FunctionalInterface`. The runtime instantiates window functions by class name, so a deployed `WindowFunction` is always a class and can override the new overload; the lambda restriction applies only to in-process/test usage.
+
+`WindowFunctionExecutor` always calls `process(Window<Record<X>>, WindowContext)`. The default overload delegates to `process(window.get(), context)` unless a class overrides the window overload.
+
+#### Empty-but-expired delivery
+
+`TimeTriggerPolicy` fires `onTrigger()` on a fixed-rate timer even after traffic stops. When the last events leave the window, `WindowManager` already drains them into `expired`, then skips `onActivation` because `events` is empty. `WindowFunctionExecutor.onExpiry` is a no-op, so those expirations never reach user code. `WindowManagerTest.testTimeBasedWindowExpiry` asserts this today.
+
+After this PIP, `WindowManager` calls `onActivation` when `events` is empty but `expired` is not. The executor still calls `process(Window, …)`:
+
+- A class that overrides `process(Window, …)` receives `get()` empty and `getExpired()` populated, so it can apply the eviction (for example decrement a running sum).
+- A class that does not override it hits the default method, which sees empty `get()` and returns `null`. The original `process(Collection, …)` is not called.
+- A bare `java.util.function.Function` is not invoked either; the executor returns `null` when `get()` is empty.
+
+### Design tradeoff
+
+Keeping `process(Collection, …)` as the single abstract method is what makes the change source- and binary-compatible. The cost is that an incremental implementation must still declare a `process(Collection, …)` body. The runtime never calls that body unless `process(Window, …)` delegates to it.
+
+That collection method should contain the full-window computation. `process(Window, …)` should call it when incremental state is missing (first activation after construction or restart). When a class overrides `process(Window, …)`, that override is the only runtime entry point.
+
+Data flow (after the change):
+
+```mermaid
+flowchart TD
+    WM["WindowManager.onTrigger()"]
+    WL["onActivation(tuples, newTuples, expiredTuples)"]
+    WE["WindowFunctionExecutor.processWindow()"]
+    WI["WindowImpl(tuples, newTuples, expiredTuples)"]
+
+    subgraph executor ["process(Window, WindowContext) dispatch"]
+        P1["WindowFunction class override<br/>→ process(window, context)"]
+        P2["WindowFunction default<br/>→ process(get()) if non-empty, else null"]
+        P3["bareWindowFunction<br/>→ apply(values) if get() non-empty"]
+    end
+
+    WM --> WL --> WE --> WI --> executor
+```
+
+# Detailed Design
+
+## Design & Implementation Details
+
+### Change 1: Move `Window<T>` to `api-java`
+
+**Current path**: `pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/Window.java`
+
+**New path**: `pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Window.java`
+
+The interface methods are unchanged in signature; the package moves to the public API module :
+
+```java
+// pulsar-functions/api-java/.../api/Window.java
+/**
+ * A per-trigger view of events in a window.
+ * <p>
+ * A {@code Window} instance is valid only for the duration of the
+ * {@link WindowFunction#process(Window, WindowContext)} call
+ * in which it is passed. Callers must not retain it across triggers.
+ * </p>
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface Window<T> {
+    /**
+     * All events in the current window for this activation.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     * <p>
+     * May be empty when this activation exists only to report expirations
+     * (see {@link #getExpired()}).
+     * </p>
+     */
+    List<T> get();
+
+    /**
+     * Events added to the window since the previous trigger.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     */
+    List<T> getNew();
+
+    /**
+     * Events removed from the window since the previous trigger.
+     * <p>
+     * The returned list is a snapshot for the current {@code process()} invocation only.
+     * Callers must not modify it. Mutation does not affect the runtime's live
+     * window state or the records the executor will ack.
+     * </p>
+     * <p>
+     * When the last events leave the window, {@link #get()} may be empty while
+     * this list is not.
+     * </p>
+     */
+    List<T> getExpired();
+
+    /**
+     * The window end timestamp for this activation.
+     * <p>
+     * Populated by {@code WindowFunctionExecutor} when it constructs the
+     * {@code WindowImpl} for each trigger. {@code WindowImpl} stores the value
+     * as supplied; it does not derive timestamps itself.
+     * </p>
+     * <p>
+     * For time-based windows this is the window's end boundary: the trigger
+     * policy's reference time (watermark in event-time mode, processing time in
+     * processing-time mode). For count-based windows this is the trigger policy's
+     * reference timestamp taken at trigger time, not a window boundary, and
+     * {@link #getStartTimestamp()} is {@code null}.
+     * </p>
+     */
+    Long getEndTimestamp();
+
+    /**
+     * The window start timestamp for this activation.
+     * <p>
+     * Populated by {@code WindowFunctionExecutor} when it constructs the
+     * {@code WindowImpl} for each trigger. When a time-based window length
+     * ({@code windowLengthDurationMs}) is configured, the executor sets this to
+     * {@code getEndTimestamp() - windowLengthDurationMs}. Returns {@code null}
+     * for count-based windows (no time-based window length).
+     * </p>
+     * <p>
+     * For sliding windows, start/end bound the <em>current activation's</em>
+     * window interval (window length), not the sliding step size
+     * ({@code slidingInterval*}).
+     * </p>
+     */
+    Long getStartTimestamp();
+}
+```
+
+**Migration of `org.apache.pulsar.functions.windowing.Window`**
+
+Delete the internal interface file. `WindowImpl` implements `org.apache.pulsar.functions.api.Window` directly, and every reference in `pulsar-functions/instance` is updated to import the public type:
+
+| File | Change |
+|------|--------|
+| `api-java/.../api/Window.java` | **Added** — canonical public interface (Javadoc, lifetime/mutability contracts) |
+| `windowing/Window.java` | **Removed** |
+| `windowing/WindowImpl.java` | `implements org.apache.pulsar.functions.api.Window<T>` |
+| `windowing/WindowFunctionExecutor.java` | import `org.apache.pulsar.functions.api.Window`; snapshot the ack list before `process()` |
+| `windowing/WindowFunctionExecutorTest.java` | import `org.apache.pulsar.functions.api.Window` |
+
+The old FQCN was never part of the supported public Functions API. Within this repository, only the files above referenced it; all are migrated to `org.apache.pulsar.functions.api.Window`. No deprecated shim is shipped in Apache Pulsar itself.
+
+The lists stay mutable. `get()` currently returns the same `tuples` list that `processWindow()` later iterates to ack records under `ATLEAST_ONCE`, so `window.get().clear()` would silently skip those acks. Wrapping the getters in `Collections.unmodifiableList` would turn that into `UnsupportedOperationException` for any existing `process(Collection, …)` implementation that already mutates its input, including via the default `process(Window)` overload. Instead, `processWindow()` copies `tuples` into a separate ack list **before** calling `process()`, and acks from that copy. Mutation of `get()` / `getNew()` / `getExpired()` then cannot reach the ack path or `WindowManager`'s live event store. Javadoc still tells callers not to modify the lists; that rule is advisory, not load-bearing.
+
+### Change 2: Add a `Window<Record<X>>` default overload to `WindowFunction<X, T>`
+
+**Path**: `pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/WindowFunction.java`
+
+```java
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+@FunctionalInterface
+public interface WindowFunction<X, T> {
+    /**
+     * Process all records in the triggered window.
+     *
+     * <p>This remains the single abstract method of the interface, so existing
+     * implementations continue to work without source changes. Incremental
+     * implementations should put the full-window computation here and have
+     * {@link #process(Window, WindowContext)} fall back to it when incremental
+     * state is not available.</p>
+     *
+     * <p>The runtime does not invoke this method when the current window is
+     * empty.</p>
+     *
+     * @param input   all current records in the window
+     * @param context the window function context
+     * @return the output, or {@code null} to suppress output
+     */
+    T process(Collection<Record<X>> input, WindowContext context) throws Exception;
+
+    /**
+     * Process the triggered window with access to incremental window views.
+     *
+     * <p>The default implementation preserves existing behavior: if
+     * {@link Window#get()} is empty it returns {@code null} without calling
+     * {@link #process(Collection, WindowContext)}; otherwise it delegates with
+     * {@link Window#get()}.</p>
+     *
+     * <p>Implementations can override this method to access newly added events
+     * ({@link Window#getNew()}) and expired events ({@link Window#getExpired()}).
+     * When this method is overridden, it is the runtime entry point; the
+     * collection-based method is invoked only if this override delegates to it.</p>
+     *
+     * @param window  the window view for this activation, valid only for the
+     *                duration of this method call. Callers must not retain the
+     *                {@code Window} reference or any list returned by its getters
+     *                across invocations.
+     * @param context     the window function context
+     * @return the output, or {@code null} to suppress output
+     */
+    default T process(Window<Record<X>> window, WindowContext context) throws Exception {
+        List<Record<X>> current = window.get();

Review Comment:
   **[NIT] The default-method body needs a java.util.List import**
   
   [`WindowFunction.java:21`](https://github.com/apache/pulsar/blob/26e26b7464b198d5af5dd5f9f52842834d15242b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/WindowFunction.java#L21) imports only `java.util.Collection`. The new body uses `List<Record<X>>`, so pasting it as shown does not compile. Either mention the import or write `Collection<Record<X>> current = window.get();`, which is what the example already does.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org