diff --git a/src/main/java/io/confluent/flink/examples/table/Example_12_StatefulProcessTableFunction.java b/src/main/java/io/confluent/flink/examples/table/Example_12_StatefulProcessTableFunction.java new file mode 100644 index 0000000..d931078 --- /dev/null +++ b/src/main/java/io/confluent/flink/examples/table/Example_12_StatefulProcessTableFunction.java @@ -0,0 +1,272 @@ +package io.confluent.flink.examples.table; + +import io.confluent.flink.plugin.ConfluentSettings; +import io.confluent.flink.plugin.ConfluentTableDescriptor; +import io.confluent.flink.plugin.ConfluentTools; +import io.confluent.flink.plugin.StatementHandle; + +import org.apache.flink.table.annotation.ArgumentHint; +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.annotation.StateHint; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.functions.ProcessTableFunction; +import org.apache.flink.types.Row; + +import java.util.List; + +import static org.apache.flink.table.annotation.ArgumentTrait.SET_SEMANTIC_TABLE; +import static org.apache.flink.table.api.Expressions.$; +import static org.apache.flink.table.api.Expressions.row; + +/** + * A table program example illustrating the stateful side of a {@link ProcessTableFunction} + * (PTF) and how its state survives a full stop/resume lifecycle on Confluent Cloud. + * + *

Where {@link Example_11_ProcessTableFunction} focuses on the general look-and-feel, this + * example focuses purely on keyed state: a PTF that keeps a running sum per partition key. + * The point of the example is to show that this accumulated state is durable. When the statement is + * stopped and later resumed, Flink restores the state and the running sums continue exactly where + * they left off. There is no data loss (rows produced while the statement was stopped are still + * processed) and no double counting (the sums are not restarted from zero). + * + *

Along the way it demonstrates the Confluent lifecycle tooling that makes this observable: + * + *

+ * + *

Background: state in Process Table Functions

+ * + *

A few concepts from Apache Flink are worth keeping in mind while reading this example: + * + *

+ * + *

Prerequisites

+ * + *

Unlike the read-only examples, this program creates tables and submits a long-running + * statement in your environment. Point {@link #TARGET_CATALOG} and {@link #TARGET_DATABASE} at + * a catalog (environment) and database (Kafka cluster) you have write access to. The program cleans + * up after itself: it deletes the statement and drops the tables it created before returning. + */ +public class Example_12_StatefulProcessTableFunction { + + // Fill this with an environment you have write access to + static final String TARGET_CATALOG = ""; + + // Fill this with a Kafka cluster you have write access to + static final String TARGET_DATABASE = ""; + + // Names of the objects this example creates. They are dropped again at the end. + static final String SOURCE_TABLE = "MyExampleSourceTable"; + static final String SINK_TABLE = "MyExampleSinkTable"; + + // A stable, human-readable statement name; so we can address the running pipeline. Statement + // names must be unique within an environment/region for an organization. + static final String STATEMENT_NAME = "example-12-stateful-ptf"; + + // 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) throws Exception { + // Setup connection properties to Confluent Cloud + EnvironmentSettings settings = ConfluentSettings.fromResource("/cloud.properties"); + + // Initialize the session context to get started + TableEnvironment env = TableEnvironment.create(settings); + + // Set default catalog and database + env.useCatalog(TARGET_CATALOG); + env.useDatabase(TARGET_DATABASE); + + // Start from a clean slate so the example is deterministic and re-runnable. + env.dropTable(SOURCE_TABLE); + env.dropTable(SINK_TABLE); + + // Create a source and a sink table with the Table Descriptor API. + // ConfluentTableDescriptor.forManaged() targets Confluent's managed connector. + // Both tables share the same schema and use a single bucket to keep the output ordering + // easy to reason about. + Schema schema = + Schema.newBuilder() + .column("name", DataTypes.STRING().notNull()) + .column("score", DataTypes.INT().notNull()) + .build(); + ConfluentTableDescriptor tableDescriptor = + ConfluentTableDescriptor.forManaged().schema(schema).distributedInto(1).build(); + env.createTable(SOURCE_TABLE, tableDescriptor); + env.createTable(SINK_TABLE, tableDescriptor); + + try { + runStatefulPtfLifecycle(env); + } finally { + // Clean up the tables regardless of outcome. + env.dropTable(SOURCE_TABLE); + env.dropTable(SINK_TABLE); + } + } + + private static void runStatefulPtfLifecycle(TableEnvironment env) throws Exception { + // 1. Seed the source with an initial batch of scores + // Two rows for Bob, one each for Alice and Charley. + // Because "name" partitions the PTF, each name accumulates its scores independently. + System.out.println("Inserting the initial batch of scores..."); + env.fromValues(row("Bob", 1), row("Bob", 2), row("Alice", 3), row("Charley", 4)) + .insertInto(SOURCE_TABLE) + .execute() + .await(); + + // 2. Submit the main stateful PTF pipeline + // Name the statement before submission so we can address it later. + ConfluentTools.setStatementName(env, STATEMENT_NAME); + + // The PTF reads the source as a set-semantic table, partitioned by name, and writes a + // running sum per name into the sink. This statement runs continuously on Confluent Cloud. + System.out.println("Submitting the stateful ProcessTableFunction pipeline..."); + TableResult result = + env.from(SOURCE_TABLE) + .partitionBy($("name")) + .process(ScoreAccumulator.class) + .insertInto(SINK_TABLE) + .execute(); + + // Get a handle to control the statement lifecycle. + // The handle is the primary way to stop, resume, and delete + // a running statement from Table API code. + StatementHandle handle = ConfluentTools.getStatementHandle(result); + System.out.println("Submitted statement: " + handle.getName()); + + // 3. Observe the accumulated state before stopping + // Read the first four sink rows. + // collectChangelog() starts a separate bounded read against + // the sink and returns once the requested number of rows has arrived. + // The running sums are: + // Bob: 1 -> 1, then +2 -> 3 + // Alice: 3 -> 3 + // Charley: 4 -> 4 + System.out.println("Collecting output before stop..."); + List beforeStop = + ConfluentTools.collectChangelog( + env.from(SINK_TABLE).select($("name"), $("score")), 4); + beforeStop.forEach(System.out::println); + // Expected (in any order): + // +I[Bob, 1] + // +I[Bob, 3] + // +I[Alice, 3] + // +I[Charley, 4] + + // 4. Stop the statement + // stop() blocks until the statement reaches a stopped state. + // Flink takes a snapshot of the PTF state on the way down, + // so the accumulated sums are preserved. + System.out.println("Stopping the statement (state is snapshotted)..."); + handle.stop(); + + // 5. Produce more data while the statement is stopped + // These rows land in the source topic but are not processed yet. They demonstrate that a + // stopped statement loses no input: the data waits until the statement resumes. + System.out.println("Inserting more scores while stopped..."); + env.fromValues(row("Bob", 5), row("Alice", 6)).insertInto(SOURCE_TABLE).execute().await(); + + // 6. Resume the statement + // resume() restores the previously snapshotted state and continues from where it left off. + // The accumulators are NOT reset: Bob resumes from 3, Alice from 3. + System.out.println("Resuming the statement (state is restored)..."); + handle.resume(); + + // Add one more row after the resume to show processing continues normally. + System.out.println("Inserting a final score after resume..."); + env.fromValues(row("Charley", 1)).insertInto(SOURCE_TABLE).execute().await(); + + // 7. Observe the accumulated state after restore + // Read all seven sink rows produced across the statement's lifetime. + // The post-resume rows continue the running sums rather than restarting them: + // Bob: 3 (restored) + 5 -> 8 + // Alice: 3 (restored) + 6 -> 9 + // Charley: 4 (restored) + 1 -> 5 + System.out.println("Collecting output after restore..."); + List afterRestore = + ConfluentTools.collectChangelog( + env.from(SINK_TABLE).select($("name"), $("score")), 7); + afterRestore.forEach(System.out::println); + // Expected (in any order): + // +I[Bob, 1] + // +I[Bob, 3] + // +I[Bob, 8] <- continued from the restored sum of 3, not from 5 + // +I[Alice, 3] + // +I[Alice, 9] <- continued from the restored sum of 3, not from 6 + // +I[Charley, 4] + // +I[Charley, 5] <- continued from the restored sum of 4, not from 1 + + // 8. Inspect and delete the statement + // The handle also exposes diagnostics: any warnings attached to the statement + // and the raw OpenAPI representation for deeper inspection. + handle.getWarnings() + .forEach( + w -> + System.out.println( + "Warning: " + w.getSeverity() + " " + w.getMessage())); + System.out.println("Statement phase: " + handle.getSqlV1Statement().getStatus().getPhase()); + + // delete() removes the statement entirely from the system, discarding its state. This is + // different from stop(): a deleted statement cannot be resumed. It blocks until the + // deletion completes. + System.out.println("Deleting the statement..."); + handle.delete(); + } + + /** + * A stateful {@link ProcessTableFunction} that accumulates a running sum of scores per + * partition key. + * + *

The output type is declared with a {@link DataTypeHint} so it maps onto the sink's {@code + * score} column. The framework automatically prepends the {@code name} partition key to the + * emitted row, producing the {@code (name, score)} shape the sink expects. + */ + @DataTypeHint("ROW<`score` INT NOT NULL>") + public static class ScoreAccumulator extends ProcessTableFunction { + + /** + * Per-key state. This POJO is the single value-state entry for each partition. It is small + * and fixed-size, so plain value state (read on entry, written back on exit) is the right + * choice; large or unbounded state would call for a TTL instead. + */ + public static class SumState { + public int sum = 0; + } + + /** + * Called for every input row. State parameters annotated with {@link StateHint} come first + * (after an optional {@code Context}), then the set-semantic table argument. Flink hands us + * the current state for the row's partition key; we mutate it and it is persisted + * transparently. + * + * @param state per-key running sum, scoped to the table's partition key + * @param input the current input row from the set-semantic table + */ + public void eval(@StateHint SumState state, @ArgumentHint(SET_SEMANTIC_TABLE) Row input) { + Integer score = input.getFieldAs("score"); + state.sum += score; + collect(Row.of(state.sum)); + } + } +}