Restructure examples into interactive/app/functions; add PTF examples

.env.template

@@ -0,0 +1,30 @@
+# Local configuration for running the examples: connection settings + secrets, provided as
+# environment variables. This is the single place to configure a local run; it is loaded by
+# bin/run.sh and bin/jshell.sh, and mirrors the variables a deployment platform injects in production.
+#
+# Setup:
+#   cp .env.template .env      # then fill in real values (.env is git-ignored)
+#
+# To load it into a plain shell instead of via the scripts:
+#   set -a; . ./.env; set +a   # (or: export $(grep -v '^#' .env | xargs), or a tool like direnv)
+
+# --- Connection configuration from .properties resource (safe, non-secret) ---
+# Where to find these values: README "Configure Table API Connection".
+# Exact names, formats, and CLI equivalents: README "Configuration Options". All five are required.
+FLINK_PROPERTIES=./src/main/resources/cloud.properties
+
+
+# --- Alternative: specify via envvars directly ---
+# Instead of the five variables above, you can keep them in a properties file and point FLINK_PROPERTIES at
+# it. See README "Via Properties File".
+# CLOUD_PROVIDER=
+# CLOUD_REGION=
+# ORG_ID=
+# ENV_ID=
+# COMPUTE_POOL_ID=
+
+# --- Secrets (keep these out of any committed file) ---
+# A single global (Cloud) API key covers both Flink access and artifact uploads.
+# See README "Authentication" for all options.
+GLOBAL_API_KEY=
+GLOBAL_API_SECRET=

.github/workflows-examples/deploy.yml

@@ -1,6 +1,6 @@
 # EXAMPLE WORKFLOW - copy this file to .github/workflows/ in your own repository.
 # It lives in workflows-examples/ because this examples repository does not deploy to a
-# real Confluent Cloud environment. It tests, deploys Example_08_IntegrationAndDeployment,
+# real Confluent Cloud environment. It tests, deploys ReferenceApp_01_IntegrationAndDeployment,
 # then lists the result. See the README's CI/CD section for the deployment model and the
 # required secrets.
 name: Deploy Table API Program
@@ -58,7 +58,7 @@ jobs:
         # pipeline restarts from its configured source offsets without resuming prior state
         # (see the README CI/CD section).
         run: >
-          java -cp target/flink-table-api-java-examples-1.0.jar io.confluent.flink.examples.table.Example_08_IntegrationAndDeployment
+          java -cp target/flink-table-api-java-examples-1.0.jar io.confluent.flink.examples.app.ReferenceApp_01_IntegrationAndDeployment
           --statement-name "$STATEMENT_NAME"
           --application-name "$APPLICATION_NAME"
           --sql.current-catalog "$TARGET_CATALOG"
@@ -69,6 +69,6 @@ jobs:
         # Submitting a background statement returns once Confluent Cloud accepts it, so
         # list reports the statement and its phase for visibility in the log.
         run: >
-          java -cp target/flink-table-api-java-examples-1.0.jar io.confluent.flink.examples.table.Example_08_IntegrationAndDeployment
+          java -cp target/flink-table-api-java-examples-1.0.jar io.confluent.flink.examples.app.ReferenceApp_01_IntegrationAndDeployment
           list
           --application-name "$APPLICATION_NAME"

.github/workflows-examples/manage.yml

@@ -66,11 +66,11 @@ jobs:
           APPLICATION_NAME: ${{ inputs.application-name }}
           STATEMENT_NAME: ${{ inputs.statement-name }}
         # --statement-name is added only when set (list then covers all statements).
-        # --action.await makes stop/resume/delete block until the target phase so the exit
-        # code reflects the outcome; list and describe ignore it.
+        # --action.await makes stop/resume/delete block until the target phase so the exit code
+        # reflects the outcome; list and describe ignore it.
         run: |
           args=("$ACTION" --application-name "$APPLICATION_NAME" --action.await)
           if [ -n "$STATEMENT_NAME" ]; then
             args+=(--statement-name "$STATEMENT_NAME")
           fi
-          java -cp target/flink-table-api-java-examples-1.0.jar io.confluent.flink.examples.table.Example_08_IntegrationAndDeployment "${args[@]}"
+          java -cp target/flink-table-api-java-examples-1.0.jar io.confluent.flink.examples.app.ReferenceApp_01_IntegrationAndDeployment "${args[@]}"

.gitignore

@@ -6,3 +6,4 @@ target/
 cloud.properties
 dependency-reduced-pom.xml
 *.env
+.envrc

bin/jshell.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+#
+# Start a JShell session with the examples on the classpath and the connection
+# config + secrets loaded from .env, so you can explore the Table API live.
+#
+# Usage:
+#   ./bin/jshell.sh
+set -euo pipefail
+
+# Always operate from the project root, regardless of where the script is invoked.
+cd "$(dirname "$0")/.."
+
+# Load .env (if present) so FLINK_PROPERTIES and the secret variables reach the JVM.
+if [ -f .env ]; then
+    set -a
+    # shellcheck disable=SC1091
+    . ./.env
+    set +a
+fi
+
+JAR="target/flink-table-api-java-examples-1.0.jar"
+if [ ! -f "$JAR" ]; then
+    echo "Building $JAR ..."
+    ./mvnw -q clean package -DskipTests
+fi
+
+exec jshell --class-path "$JAR" --startup ./jshell-init.jsh

bin/run.sh

@@ -0,0 +1,62 @@
+#!/usr/bin/env bash
+#
+# Run a Table API example locally.
+#
+# Loads connection config + secrets from .env (git-ignored) into the environment
+# so the Confluent plugin picks up FLINK_PROPERTIES and the secret variables,
+# builds the shaded jar if it is missing, then runs the chosen example class.
+#
+# Usage:
+#   ./bin/run.sh [ExampleClass] [program args...]
+#
+# Examples:
+#   ./bin/run.sh Example_00_HelloWorld
+#   ./bin/run.sh ReferenceApp_01_IntegrationAndDeployment --statement-name demo --on-conflict replace
+set -euo pipefail
+
+# Always operate from the project root, regardless of where the script is invoked.
+cd "$(dirname "$0")/.."
+
+# Load .env (if present) so FLINK_PROPERTIES and the secret variables reach the JVM.
+if [ -f .env ]; then
+    set -a
+    # shellcheck disable=SC1091
+    . ./.env
+    set +a
+fi
+
+EXAMPLE="${1:-ReferenceApp_01_IntegrationAndDeployment}"
+[ $# -gt 0 ] && shift
+
+JAR="target/flink-table-api-java-examples-1.0.jar"
+if [ ! -f "$JAR" ]; then
+    echo "Building $JAR ..."
+    ./mvnw -q clean package -DskipTests
+fi
+
+# Examples live in two packages -- interactive/ (run inline and print) and app/ (deployable
+# statements) -- plus TableProgramTemplate at the top level. Resolve the fully-qualified name from
+# the jar so callers pass only the class name. The name is treated as a prefix, so "Example_02"
+# resolves to "Example_02_UnboundedTables". Inner classes (those with a '$') are ignored.
+# '|| true' keeps a no-match (unzip exits 11, grep -v exits 1) from tripping 'set -e' here, so the
+# empty-result check below can report it instead of the script aborting silently.
+MATCHES=$(unzip -l "$JAR" "io/confluent/flink/examples/$EXAMPLE*.class" \
+        "io/confluent/flink/examples/*/$EXAMPLE*.class" 2>/dev/null \
+    | awk '/\.class$/ {print $4}' \
+    | grep -v '\$' \
+    | sed 's#/#.#g; s#\.class$##' \
+    | sort -u || true)
+
+if [ -z "$MATCHES" ]; then
+    echo "Could not find an example class matching '$EXAMPLE' in $JAR" >&2
+    exit 1
+fi
+if [ "$(printf '%s\n' "$MATCHES" | wc -l)" -gt 1 ]; then
+    echo "'$EXAMPLE' is ambiguous; matches:" >&2
+    printf '%s\n' "$MATCHES" | sed 's#.*\.#  #' >&2
+    echo "Please use a more specific name." >&2
+    exit 1
+fi
+CLASS="$MATCHES"
+
+exec java -cp "$JAR" "$CLASS" "$@"

pom.xml

@@ -83,6 +83,7 @@
             <version>${junit.version}</version>
             <scope>test</scope>
         </dependency>
+
         <dependency>
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>
@@ -232,7 +233,7 @@
                                         implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                                 <transformer
                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-                                    <mainClass>io.confluent.flink.examples.table.Example_00_HelloWorld</mainClass>
+                                    <mainClass>io.confluent.flink.examples.interactive.Example_00_HelloWorld</mainClass>
                                 </transformer>
                             </transformers>
                         </configuration>

src/main/java/io/confluent/flink/examples/table/TableProgramTemplate.java → src/main/java/io/confluent/flink/examples/TableProgramTemplate.java

@@ -1,4 +1,4 @@
-package io.confluent.flink.examples.table;
+package io.confluent.flink.examples;
 
 // spotless:off
 
@@ -54,7 +54,9 @@ public class TableProgramTemplate {
     public static void main(String[] args) {
         EnvironmentSettings settings =
                 ConfluentSettings.newBuilderFromResource("/cloud.properties")
+                        .setApplicationName("my-table-program")
                         .setOption("sql.local-time-zone", "UTC")
+                        .applyArgs(args)
                         .build();
 
         TableEnvironment env = TableEnvironment.create(settings);

src/main/java/io/confluent/flink/examples/app/ReferenceApp_01_IntegrationAndDeployment.java

@@ -0,0 +1,131 @@
+package io.confluent.flink.examples.app;
+
+import io.confluent.flink.plugin.ConfluentSettings;
+import io.confluent.flink.plugin.ConfluentTableDescriptor;
+import io.confluent.flink.plugin.ConfluentTools;
+
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.lit;
+
+/**
+ * An example that illustrates how to structure, test, and deploy a table program for production use
+ * in a CI/CD pipeline.
+ *
+ * <p>The example separates two concerns that often end up entangled:
+ *
+ * <ul>
+ *   <li>The pipeline logic in {@link VendorsPerBrand} is plain Table API code without any
+ *       Confluent-specific dependencies. It receives its input table as a parameter instead of
+ *       resolving it from a catalog. This makes the logic executable on Apache Flink in local unit
+ *       tests, where the input is mocked with {@code fromValues()} (see {@code
+ *       ReferenceApp_01_IntegrationAndDeploymentTest}), as well as on Confluent Cloud for Apache
+ *       Flink, where the input is a Kafka-backed table (see {@code
+ *       ReferenceApp_01_IntegrationAndDeploymentIT}).
+ *   <li>The {@link #main(String[])} method is the deployment entrypoint. It wires the pipeline to
+ *       Confluent Cloud and submits it as a long-running background statement.
+ * </ul>
+ *
+ * <p>The program is configured with {@code ConfluentSettings.newBuilder().applyArgs(args)}: the
+ * builder starts from the environment (a {@code FLINK_PROPERTIES} properties file plus environment
+ * variables for secrets), and {@link ConfluentSettings.Builder#applyArgs(String[])} layers the
+ * per-deployment command-line arguments on top. This also enables the plugin's built-in CI/CD
+ * lifecycle actions: when the JAR is run with an action as the first argument ({@code list}, {@code
+ * describe}, {@code resume}, {@code stop}, or {@code delete}), the plugin executes the action and
+ * exits before the deployment logic runs, so the same JAR both deploys and manages (see {@code
+ * .github/workflows-examples/manage.yml}).
+ *
+ * <p>The application name is set in code as a default and can be overridden per deployment with
+ * {@code --application-name}. The statement name is deployment configuration passed via {@code
+ * --statement-name}, which the lifecycle actions use to target the right statement for both deploy
+ * and management. A program that submits several statements instead names each one in code via
+ * {@link ConfluentTools#setStatementName(TableEnvironment, String)}.
+ *
+ * <p>Re-running the deployment with unchanged code is idempotent. When the pipeline changed, pass
+ * {@code --on-conflict replace} to replace the existing statement; see the README's CI/CD section
+ * for what that means for stateful pipelines. The README also covers configuration, environment
+ * promotion, and the full set of workflow steps.
+ *
+ * <p>NOTE: This example requires write access to a Kafka cluster, selected with the {@code
+ * sql.current-catalog} (environment name) and {@code sql.current-database} (Kafka cluster name)
+ * configuration options.
+ *
+ * <p>ALSO NOTE: The example submits an unbounded background statement. Use the lifecycle actions
+ * (see {@code .github/workflows-examples/manage.yml}) or the Web UI to stop and delete the
+ * statement afterward to clean up resources.
+ */
+public class ReferenceApp_01_IntegrationAndDeployment {
+
+    private static final Logger LOG =
+            LoggerFactory.getLogger(ReferenceApp_01_IntegrationAndDeployment.class);
+
+    // Name of the table that stores the results
+    static final String TARGET_TABLE = "VendorsPerBrand";
+
+    /**
+     * The pipeline logic under test: counts the number of vendors per brand.
+     *
+     * <p>This class must not reference any {@code io.confluent.flink.plugin} classes so that unit
+     * tests can run it on Apache Flink without the plugin on the classpath.
+     */
+    public static class VendorsPerBrand {
+        public static Table buildPipeline(Table products) {
+            return products.groupBy($("brand")).select($("brand"), lit(1).count().as("vendors"));
+        }
+    }
+
+    // The main() method performs the deployment, unless an action argument is present, in which
+    // case applyArgs(...) records it and build() below executes that action and exits before the
+    // rest of this method runs.
+    public static void main(String[] args) {
+        // Connection settings come from the environment (a FLINK_PROPERTIES file for safe config
+        // plus environment variables for the API key and secret). The application name is defaulted
+        // in code here. The per-deployment arguments -- the statement name, the target catalog and
+        // database (sql.current-catalog / sql.current-database), any lifecycle action, and an
+        // optional --application-name override -- are layered on top with applyArgs. In GitHub
+        // Actions the secrets map to repository or environment secrets; see the README.
+        EnvironmentSettings settings =
+                ConfluentSettings.newBuilder()
+                        .setApplicationName("vendors-per-brand")
+                        .applyArgs(args)
+                        .build();
+        TableEnvironment env = TableEnvironment.create(settings);
+
+        LOG.info("Creating table... {}", TARGET_TABLE);
+        // The pipeline owns its output table and creates it on the first deployment.
+        env.createTable(
+                TARGET_TABLE,
+                ConfluentTableDescriptor.forManaged()
+                        .schema(
+                                Schema.newBuilder()
+                                        .column("brand", DataTypes.STRING().notNull())
+                                        .column("vendors", DataTypes.BIGINT())
+                                        .primaryKey("brand")
+                                        .build())
+                        .distributedInto(1)
+                        .build(),
+                true);
+
+        LOG.info("Deploying statement...");
+        // The same pipeline logic that was tested locally and against Confluent Cloud now runs
+        // unbounded on the continuously generated rows of the examples catalog.
+        ConfluentTools.setStatementName(env, "pipeline");
+        Table products = env.from("`examples`.`marketplace`.`products`");
+        TableResult result =
+                VendorsPerBrand.buildPipeline(products).insertInto(TARGET_TABLE).execute();
+
+        // Print the final submitted name (application prefix included) for use with the lifecycle
+        // actions. If no name was configured, the plugin generates one, which is not addressable
+        // for later management; CI/CD deployments should always pass --statement-name.
+        LOG.info("Statement has been deployed as: {}", ConfluentTools.getStatementName(result));
+    }
+}