Add an example for ProcessTableFunctions

Author: MartijnVisser

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 2026-04-15
+
+### Features and Fixes
+
+- Added Example_11: ProcessTableFunction with event-time timers, demonstrating user inactivity
+  detection on the `examples.marketplace.clicks` table.
+
 ## 2024-09-13
 
 ### Features and Fixes

src/main/java/io/confluent/flink/examples/table/Example_11_ProcessTableFunction.java

@@ -22,26 +22,20 @@
  * A table program example illustrating how to use a {@link ProcessTableFunction} (PTF) in the Flink
  * Table API.
  *
- * <p>This example demonstrates a session tracking PTF invoked against the {@code
- * examples.marketplace.clicks} table. The PTF:
+ * <p>This example detects inactive users on the {@code examples.marketplace.clicks} table. For each
+ * user, it counts clicks and registers an event-time timer. When no new click arrives within the
+ * configured timeout, the timer fires and emits an alert with the user's accumulated click count.
  *
- * <ul>
- *   <li>Starts a session on the first click event per user
- *   <li>Extends the session timeout on each subsequent click
- *   <li>Automatically emits a session summary when the timeout expires
- *   <li>Tracks click counts and URLs within each session
- * </ul>
- *
- * <p>Sessions are partitioned by {@code user_id} and end via timeout when no new clicks arrive
- * within the configured period.
+ * <p>Unlike a windowed aggregation, the PTF emits exactly once per inactivity period, combining
+ * running state with the absence of events.
  */
 public class Example_11_ProcessTableFunction {
 
     // Fill this with an environment you have write access to
-    static final String TARGET_CATALOG = "mvisser";
+    static final String TARGET_CATALOG = "";
 
     // Fill this with a Kafka cluster you have write access to
-    static final String TARGET_DATABASE = "standard_cluster";
+    static final String TARGET_DATABASE = "";
 
     // All logic is defined in a main() method. It can run both in an IDE or CI/CD system.
     public static void main(String[] args) {
@@ -55,104 +49,58 @@ public static void main(String[] args) {
         env.useCatalog(TARGET_CATALOG);
         env.useDatabase(TARGET_DATABASE);
 
-        // Invoke the PTF inline (without registration) against the clicks table.
-        // The Table API creates a temporary JAR with all required dependencies,
-        // uploads it to Confluent Cloud, and binds the function lifecycle to the query.
-        // Sessions are partitioned by user_id and time out after 60 seconds of inactivity.
-        System.out.println("Executing SessionTracker against clicks table...");
+        // Invoke the PTF inline against the clicks table, partitioned by user_id.
+        // The function's lifecycle is bound to this query.
+        System.out.println("Executing inline ProcessTableFunction...");
         env.from("`examples`.`marketplace`.`clicks`")
                 .partitionBy($("user_id"))
                 .process(
-                        SessionTracker.class,
-                        lit(60).asArgument("sessionTimeoutSeconds"),
+                        ClickInactivityMonitor.class,
+                        lit(30).asArgument("timeoutSeconds"),
                         descriptor("$rowtime").asArgument("on_time"))
                 .execute()
                 .print();
     }
 
     /**
-     * A ProcessTableFunction that tracks user sessions based on click events.
+     * A ProcessTableFunction that detects user inactivity based on click events.
      *
-     * <p>For each user (partitioned by {@code user_id}), it maintains session state and emits a
-     * {@link SessionSummary} when the session times out due to inactivity.
+     * <p>For each user (partitioned by {@code user_id}), it counts incoming clicks and registers a
+     * named event-time timer. Each new click replaces the previous timer, resetting the inactivity
+     * clock. When the timer fires (no new clicks within the timeout), an alert is emitted.
      */
-    public static class SessionTracker extends ProcessTableFunction<SessionTracker.SessionSummary> {
-
-        /** POJO output type for session summaries. */
-        public static class SessionSummary {
-            public int userId;
-            public String status;
-            public long sessionDuration;
-            public int eventCount;
-            public String firstUrl;
-            public String lastUrl;
+    public static class ClickInactivityMonitor
+            extends ProcessTableFunction<ClickInactivityMonitor.InactivityAlert> {
+
+        /** Output POJO. The framework adds the user_id partition key and rowtime automatically. */
+        public static class InactivityAlert {
+            public int clickCount;
         }
 
-        /** State class tracking user session information. */
-        public static class SessionState {
-            public String firstUrl = null;
-            public String lastUrl = null;
-            public Long sessionStart = null;
-            public int userId = 0;
-            public int eventCount = 0;
+        /** Per-user state. */
+        public static class ClickState {
+            public int clickCount = 0;
         }
 
-        /**
-         * Evaluates incoming click events and manages session state.
-         *
-         * @param ctx Process context for timers and state management
-         * @param state Session state with 2-hour TTL
-         * @param input Click event row from the clicks table
-         * @param sessionTimeoutSeconds Configurable session timeout in seconds
-         */
         public void eval(
                 Context ctx,
-                @StateHint(ttl = "2 hours") SessionState state,
+                @StateHint ClickState state,
                 @ArgumentHint({SET_SEMANTIC_TABLE, REQUIRE_ON_TIME}) Row input,
-                Integer sessionTimeoutSeconds)
-                throws Exception {
+                Integer timeoutSeconds) {
+
+            state.clickCount++;
 
+            // Each new click pushes the timeout forward; the timer fires only after true
+            // inactivity.
             TimeContext<Instant> timeCtx = ctx.timeContext(Instant.class);
-            int userId = input.getFieldAs("user_id");
-            String url = input.getFieldAs("url");
-
-            if (state.firstUrl == null) {
-                // First click in session
-                state.firstUrl = url;
-                state.sessionStart = timeCtx.time().toEpochMilli();
-                state.userId = userId;
-                state.eventCount = 1;
-            } else {
-                // Subsequent click: increment count
-                state.eventCount++;
-            }
-
-            // Track URL and extend session timeout
-            state.lastUrl = url;
             timeCtx.registerOnTime(
-                    "timeout", timeCtx.time().plus(Duration.ofSeconds(sessionTimeoutSeconds)));
+                    "inactivity", timeCtx.time().plus(Duration.ofSeconds(timeoutSeconds)));
         }
 
-        /**
-         * Timer callback invoked when the session timeout is reached, indicating no new clicks
-         * arrived within the configured period.
-         */
-        public void onTimer(OnTimerContext onTimerCtx, SessionState state) throws Exception {
-
-            if (state.sessionStart != null) {
-                Instant currentTime = onTimerCtx.timeContext(Instant.class).time();
-                long duration = currentTime.toEpochMilli() - state.sessionStart;
-
-                SessionSummary summary = new SessionSummary();
-                summary.userId = state.userId;
-                summary.status = "TIMEOUT";
-                summary.sessionDuration = duration;
-                summary.eventCount = state.eventCount;
-                summary.firstUrl = state.firstUrl;
-                summary.lastUrl = state.lastUrl;
-
-                collect(summary);
-            }
+        public void onTimer(ClickState state) {
+            InactivityAlert alert = new InactivityAlert();
+            alert.clickCount = state.clickCount;
+            collect(alert);
         }
     }
 }