How to find the min/max in a stream of events using Flink SQL

Run it

Prerequisites

This tutorial installs Confluent Platform using Docker. Before proceeding:

• Install Docker Desktop (version 4.0.0 or later) or Docker Engine (version 19.03.0 or later) if you don’t already have it
• Install the Docker Compose plugin if you don’t already have it. This isn’t necessary if you have Docker Desktop since it includes Docker Compose.
• Start Docker if it’s not already running, either by starting Docker Desktop or, if you manage Docker Engine with systemd, via systemctl
• Verify that Docker is set up properly by ensuring no errors are output when you run docker info and docker compose version on the command line

Initialize the project

To get started, make a new directory anywhere you’d like for this project:

mkdir aggregate-minmax && cd aggregate-minmax

Get Confluent Platform and Apache Flink®

Next, create the following docker-compose.yml file to obtain Confluent Platform (for Kafka in the cloud, see Confluent Cloud) and Apache Flink®. The Docker Compose file will start three Flink® containers that have Kafka connector dependencies preinstalled: an interactive Flink SQL client (flink-sql-client) that sends streaming SQL jobs to the Flink Job Manager (flink-job-manager), which in turn assigns tasks to the Flink Task Manager (flink-task-manager) in the Flink cluster.

version: '2'
services:
  broker:
    image: confluentinc/cp-kafka:7.4.1
    hostname: broker
    container_name: broker
    ports:
    - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
      KAFKA_LISTENERS: PLAINTEXT://broker:9092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:29092
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
  schema-registry:
    image: confluentinc/cp-schema-registry:7.3.0
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
    - broker
    ports:
    - 8081:8081
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: broker:9092
  flink-sql-client:
    image: cnfldemos/flink-sql-client-kafka:1.16.0-scala_2.12-java11
    hostname: flink-sql-client
    container_name: flink-sql-client
    depends_on:
    - flink-jobmanager
    environment:
      FLINK_JOBMANAGER_HOST: flink-jobmanager
    volumes:
    - ./settings/:/settings
  flink-jobmanager:
    image: cnfldemos/flink-kafka:1.16.0-scala_2.12-java11
    hostname: flink-jobmanager
    container_name: flink-jobmanager
    ports:
    - 9081:9081
    command: jobmanager
    environment:
    - |
      FLINK_PROPERTIES=
      jobmanager.rpc.address: flink-jobmanager
      rest.bind-port: 9081
  flink-taskmanager:
    image: cnfldemos/flink-kafka:1.16.0-scala_2.12-java11
    hostname: flink-taskmanager
    container_name: flink-taskmanager
    depends_on:
    - flink-jobmanager
    command: taskmanager
    scale: 1
    environment:
    - |
      FLINK_PROPERTIES=
      jobmanager.rpc.address: flink-jobmanager
      taskmanager.numberOfTaskSlots: 10

Launch it by running:

docker compose up -d

Write the program interactively using the CLI

The best way to interact with Flink SQL when you’re learning how things work is with the Flink SQL CLI. Fire it up as follows:

docker exec -it flink-sql-client sql-client.sh

Our tutorial computes the highest grossing and lowest grossing films per year in our data set. To keep things simple, we’re going to create a table backed by a Kafka topic with annual sales data in it. In a real-world data pipeline, this would probably be the output of another table that takes a stream of individual sales events and aggregates them into annual totals, but we’ll save ourselves that trouble and just create the annual sales data directly.

This line of Flink SQL DDL creates a table and its underlying Kafka topic to represent the annual sales totals. Note that we are defining the schema for the table, which includes four fields: id, title, release_year, and total_sales. We are also specifying that the underlying Kafka topic—which Flink SQL will auto-create—be called movie-sales and have just one partition (the default num.partitions configured in the broker), and that its messages will be in Avro format.

CREATE TABLE movie_sales (
    id INT,
    title STRING,
    release_year INT,
    total_sales INT
) WITH (
    'connector' = 'kafka',
    'topic' = 'movie-sales',
    'properties.bootstrap.servers' = 'broker:9092',
    'scan.startup.mode' = 'earliest-offset',
    'key.format' = 'raw',
    'key.fields' = 'id',
    'value.format' = 'avro-confluent',
    'value.avro-confluent.url' = 'http://schema-registry:8081',
    'value.fields-include' = 'EXCEPT_KEY'
);

Confluent Cloud manages several options for you when using Flink SQL. So, if you run this tutorial on Confluent Cloud, you can copy just the CREATE TABLE statements without the WITH clause when creating tables. Consult the Flink SQL WITH documentation for the full list supported options when creating a table.

Let’s add a small amount of data to our table, so we can see our query work. Go ahead and paste this statement into the Flink SQL CLI now.

INSERT INTO movie_sales VALUES
    (0, 'Avengers: Endgame', 2019, 856980506),
    (1, 'Captain Marvel', 2019, 426829839),
    (2, 'Toy Story 4', 2019, 401486230),
    (3, 'The Lion King', 2019, 385082142),
    (4, 'Black Panther', 2018, 700059566),
    (5, 'Avengers: Infinity War', 2018, 678815482),
    (6, 'Deadpool 2', 2018, 324512774),
    (7, 'Beauty and the Beast', 2017, 517218368),
    (8, 'Wonder Woman', 2017, 412563408),
    (9, 'Star Wars Ep. VIII: The Last Jedi', 2017, 517218368);

With our test data in place, let’s try a query to compute the min and max. A SELECT statement in Flink SQL is called a continuous query because it will continue to run and produce results dynamically. This query is transient, meaning that after we stop it, it is gone and will not keep processing the input stream. Further, the results aren’t persisted anywhere. We’ll create a persistent query, the contrast to a transient push query, a few steps from now.

If you’re familiar with SQL, the text of the query itself is fairly self-explanatory. We are calculating the highest and lowest grossing movie figures by year using MIN and MAX aggregations on the total_sales column. This query will keep running, continuing to return results until you use Ctrl-C.

SELECT
    release_year,
    MIN(total_sales) AS min_total_sales,
    MAX(total_sales) AS max_total_sales
FROM movie_sales
GROUP BY release_year;

This should yield the following output:

 release_year min_total_sales max_total_sales
         2017       412563408       517218368
         2019       385082142       856980506
         2018       324512774       700059566\

Enter Q to return to the Flink SQL prompt.

Note that these results were materialized in memory and printed in a human-readable table representation because the default sql-client.execution.result-mode configuration value is 'table'. You can view non-materialized streaming results as a changelog by running SET 'sql-client.execution.result-mode' = 'changelog'; and rerunning the same query. The results will look like this:

 op release_year min_total_sales max_total_sales
 +I         2019       426829839       426829839
 +I         2017       412563408       412563408
 -U         2019       426829839       426829839
 +U         2019       401486230       426829839
 +I         2018       324512774       324512774
 -U         2019       401486230       426829839
 +U         2019       401486230       856980506
 -U         2017       412563408       412563408
 +U         2017       412563408       517218368
 -U         2018       324512774       324512774
 +U         2018       324512774       678815482
 -U         2019       401486230       856980506
 +U         2019       385082142       856980506
 -U         2018       324512774       678815482
 +U         2018       324512774       700059566

Or, as a third option, you can see streaming results non-materialized and inline in the SQL client by running SET 'sql-client.execution.result-mode' = 'tableau'; and rerunning the query once more. In this case, the results will look very similar to changelog mode results. This is because tables sourced by the Kafka connector are unbounded and can thus only be queried in streaming mode. For further reading on these Flink SQL concepts, consult the documentation on SQL client result modes and streaming vs. batch execution

Since the output of our transient query looks right, the next step is to make the query persistent. This looks exactly like the transient query, except we first create a new table with the Upsert Kafka connector and then INSERT INTO the table. We use the Upsert Kafka connector because we only care about the most recent aggregates for a given release year (the key column). The INSERT INTO statement returns to the CLI prompt right away, having created a persistent stream processing program running in the Flink cluster, continuously processing input records and updating the resulting movie_sales_by_year table.

Now go ahead and run the following two commands in your Flink SQL session:

CREATE TABLE movie_sales_by_year (
    release_year INT,
    min_total_sales INT,
    max_total_sales INT,
    PRIMARY KEY (release_year) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'movie-sales-by-year',
    'properties.bootstrap.servers' = 'broker:9092',
    'key.format' = 'raw',
    'value.format' = 'avro-confluent',
    'value.avro-confluent.url' = 'http://schema-registry:8081',
    'value.fields-include' = 'EXCEPT_KEY'
);

INSERT INTO movie_sales_by_year
SELECT
    release_year,
    MIN(total_sales) AS min_total_sales,
    MAX(total_sales) AS max_total_sales
FROM movie_sales
GROUP BY release_year;

Validate output

Seeing is believing, so let’s query the persistent movie_sales_by_year table. First, set the result mode back to table:

SET 'sql-client.execution.result-mode' = 'table';

Then query the movie_sales_by_year table:

SELECT
    release_year,
    min_total_sales,
    max_total_sales
FROM movie_sales_by_year;

This will yield the same output that the transient query did (perhaps in a different order)

 release_year min_total_sales max_total_sales
         2019       385082142       856980506
         2018       324512774       700059566
         2017       412563408       517218368

We could also query the underlying topic directly using kafka-avro-console-consumer. Open a new terminal window and run the following command:

docker exec -e SCHEMA_REGISTRY_LOG4J_OPTS=" " -it schema-registry /usr/bin/kafka-avro-console-consumer \
  --topic movie-sales-by-year \
  --from-beginning \
  --max-messages 9 \
  --timeout-ms 10000 \
  --bootstrap-server broker:9092 \
  --property key.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer \
  --property print.key=true \
  --property key.separator="-"

This will yield the following results:

2019-{"min_total_sales":{"int":856980506},"max_total_sales":{"int":856980506}}
2019-{"min_total_sales":{"int":426829839},"max_total_sales":{"int":856980506}}
2019-{"min_total_sales":{"int":401486230},"max_total_sales":{"int":856980506}}
2019-{"min_total_sales":{"int":385082142},"max_total_sales":{"int":856980506}}
2018-{"min_total_sales":{"int":700059566},"max_total_sales":{"int":700059566}}
2018-{"min_total_sales":{"int":678815482},"max_total_sales":{"int":700059566}}
2018-{"min_total_sales":{"int":324512774},"max_total_sales":{"int":700059566}}
2017-{"min_total_sales":{"int":517218368},"max_total_sales":{"int":517218368}}
2017-{"min_total_sales":{"int":412563408},"max_total_sales":{"int":517218368}}
Processed a total of 9 messages

Observe that the latest value for each key (release year) matches the final min / max aggregate values that we expect. At this point, we could also adjust the movie-sales-by-year topic’s cleanup policy to compact so that only the latest aggregates per year are retained:

docker exec -it broker /usr/bin/kafka-configs \
  --bootstrap-server localhost:29092 \
  --entity-type topics \
  --entity-name movie-sales-by-year \
  --alter \
  --add-config cleanup.policy=compact

Kafka log cleanup happens asynchronously, so this wouldn’t immediately reduce the topic down to 3 retained messages (one per year). It’s included here to show a common Kafka topic configuration used in conjunction with Flink SQL tables sourced by the Upsert Kafka connector, particularly for streaming data sets with many updates per key.

Test it

Decide what testing tools to use

Now that you have manually developed and tested your Flink SQL application, how might you create an automated test for it so that it’s easier to maintain and upgrade over time? Imagine how painful it would be to have to manually test every change or software dependency upgrade of your application, and then imagine having to do this for many applications. The benefits of automated testing are clear, but how do we get there?

First, what do we want in an automated integration test? For starters:

Real services (as opposed to mock) that our application depends on
Small resource footprint so that developers can run the test locally
Low enough latency so that development iterations aren’t hampered — not as low latency as is required for a unit test, but test duration should be on the order of seconds
Isolation so that many tests can run concurrently on the same machine when this test gets run on a build automation server, e.g., no hard-coded ports

Luckily, tools are at our disposal to solve these problems. We’ll use Testcontainers to run containerized Kafka and Schema Registry servers on dynamic ports, Flink’s support for local execution environments so that we don’t need to spin up a Flink cluster, and Flink’s Table API to execute the Flink SQL statements that comprise our application.

Create the test skeleton

The primary choices for programming language in which to write our tests are Java and Python given the need for Flink’s Table API. We’ll write ours in Java.

To start our test project, make new directories for test source code and resources within the same aggregate-minmax folder that you created earlier:

mkdir -p src/test/java/io/confluent/developer
mkdir src/test/resources

Create the following Gradle build file, named build.gradle, in the aggregate-minmax directory.

buildscript {
    repositories {
        mavenCentral()
    }
}

plugins {
    id "java"
}

sourceCompatibility = JavaVersion.VERSION_11
targetCompatibility = JavaVersion.VERSION_11
version = "0.0.1"

repositories {
    mavenCentral()
}

dependencies {
    testImplementation "com.google.guava:guava:31.1-jre"
    testImplementation "junit:junit:4.13.2"
    testImplementation 'org.testcontainers:testcontainers:1.17.6'
    testImplementation 'org.testcontainers:kafka:1.17.6'
    testImplementation "org.apache.flink:flink-sql-connector-kafka:1.16.1"
    testImplementation "org.apache.flink:flink-sql-avro-confluent-registry:1.16.1"
    testImplementation "org.apache.flink:flink-test-utils:1.16.1"
    testImplementation "org.apache.flink:flink-test-utils-junit:1.16.1"
    testImplementation "org.apache.flink:flink-table-api-java-bridge:1.16.1"
    testImplementation "org.apache.flink:flink-table-planner_2.12:1.16.1"
    testImplementation "org.apache.flink:flink-table-planner_2.12:1.16.1:tests"
    testImplementation "org.apache.flink:flink-statebackend-rocksdb:1.16.1"
}

There are a couple of important points to note in the Gradle build file:

Java sourceCompatibility and targetCompatibility are set to Java 11. Flink supports Java 8 (deprecated) and 11 as of the writing of this tutorial
The dependencies section declares test dependencies for Testcontainers and Flink. Among the handful of Flink dependencies are ones providing local execution (e.g., flink-statebackend-rocksdb), the Table API (flink-table-api-java-bridge), and Kafka connectors that can use Schema Registry (flink-sql-connector-kafka and flink-sql-avro-confluent-registry).

And be sure to run the following command to obtain the Gradle wrapper:

gradle wrapper

Create SQL resources

We could always inline the SQL statements in our Java test code, but creating separate resource files makes our test more readable and easier to maintain. Further, we can imagine parametrizing URLs as well so that we can have a single set of source-controlled queries to use in tests as well as staging or production environments.

There are a handful of resources to create for our test. These mirror the queries that we developed earlier.

Create the following file at src/test/resources/create-movie-sales.sql.template. Note the KAFKA_PORT and SCHEMA_REGISTRY_PORT placeholders in this file. Our test will dynamically assign these to the ports that Testcontainers assigns.

CREATE TABLE movie_sales (
    id INT,
    title STRING,
    release_year INT,
    total_sales INT
) WITH (
    'connector' = 'kafka',
    'topic' = 'movie-sales',
    'properties.bootstrap.servers' = 'localhost:KAFKA_PORT',
    'scan.startup.mode' = 'earliest-offset',
    'key.format' = 'raw',
    'key.fields' = 'id',
    'value.format' = 'avro-confluent',
    'value.avro-confluent.url' = 'http://localhost:SCHEMA_REGISTRY_PORT',
    'value.fields-include' = 'EXCEPT_KEY'
);

Create the following file at src/test/resources/populate-movie-sales.sql.

INSERT INTO movie_sales
VALUES (0, 'Avengers: Endgame', 2019, 856980506),
       (1, 'Captain Marvel', 2019, 426829839),
       (2, 'Toy Story 4', 2019, 401486230),
       (3, 'The Lion King', 2019, 385082142),
       (4, 'Black Panther', 2018, 700059566),
       (5, 'Avengers: Infinity War', 2018, 678815482),
       (6, 'Deadpool 2', 2018, 324512774),
       (7, 'Beauty and the Beast', 2017, 517218368),
       (8, 'Wonder Woman', 2017, 412563408),
       (9, 'Star Wars Ep. VIII: The Last Jedi', 2017, 517218368);

Create the following file at src/test/resources/create-movie-sales-by-year.sql.template. Again, note the KAFKA_PORT and SCHEMA_REGISTRY_PORT placeholders since this table uses the Upsert Kafka connector and hence must be able to communicate with Kafka and Schema Registry.

CREATE TABLE movie_sales_by_year (
    release_year INT,
    min_total_sales INT,
    max_total_sales INT,
    PRIMARY KEY (release_year) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'movie-sales-by-year',
    'properties.bootstrap.servers' = 'localhost:KAFKA_PORT',
    'key.format' = 'raw',
    'value.format' = 'avro-confluent',
    'value.avro-confluent.url' = 'http://localhost:SCHEMA_REGISTRY_PORT',
    'value.fields-include' = 'EXCEPT_KEY'
);

Create the following file at src/test/resources/populate-movie-sales-by-year.sql:

INSERT INTO movie_sales_by_year
SELECT
    release_year,
    MIN(total_sales) AS min_total_sales,
    MAX(total_sales) AS max_total_sales
FROM movie_sales
GROUP BY release_year;

Next, create the following file at src/test/resources/query-movie-sales-by-year.sql:

SELECT * FROM movie_sales_by_year;

Finally, create the following file at src/test/resources/expected-movie-sales-by-year.txt that contains our test’s expected output:

+----+--------------+-----------------+-----------------+
| op | release_year | min_total_sales | max_total_sales |
+----+--------------+-----------------+-----------------+
| +I |         2017 |       517218368 |       517218368 |
| -U |         2017 |       517218368 |       517218368 |
| +U |         2017 |       412563408 |       517218368 |
| +I |         2019 |       856980506 |       856980506 |
| -U |         2019 |       856980506 |       856980506 |
| +U |         2019 |       426829839 |       856980506 |
| -U |         2019 |       426829839 |       856980506 |
| +U |         2019 |       401486230 |       856980506 |
| -U |         2019 |       401486230 |       856980506 |
| +U |         2019 |       385082142 |       856980506 |
| +I |         2018 |       700059566 |       700059566 |
| -U |         2018 |       700059566 |       700059566 |
| +U |         2018 |       678815482 |       700059566 |
| -U |         2018 |       678815482 |       700059566 |
| +U |         2018 |       324512774 |       700059566 |

Write a test

Create the following abstract test class at src/test/java/io/confluent/developer/AbstractFlinkKafkaTest.java:

package io.confluent.developer;


import com.google.common.io.Resources;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.client.JobCancellationException;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.assertj.core.util.Sets;
import org.junit.BeforeClass;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.utility.DockerImageName;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Optional;
import java.util.Set;

import static org.testcontainers.containers.KafkaContainer.KAFKA_PORT;

/**
 * Base class for Flink SQL integration tests that use Flink's Kafka connectors. Encapsulates
 * Kafka broker and Schema Registry Testcontainer management and includes utility methods for
 * dynamically configuring Flink SQL Kafka connectors and processing Table API results.
 */
public class AbstractFlinkKafkaTest {

  protected static StreamTableEnvironment streamTableEnv;
  protected static Integer schemaRegistryPort, kafkaPort;

  @BeforeClass
  public static void setup() {
    // create Flink table environment that test subclasses will use to execute SQL statements
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(4);
    env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
    env.setStateBackend(new EmbeddedRocksDBStateBackend());
    streamTableEnv = StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().inStreamingMode().build());


    // Start Kafka and Schema Registry Testcontainers. Set the exposed ports that test subclasses
    // can use to dynamically configure Kafka connectors. Schema Registry enables connectors to
    // be configured with 'value.format' = 'avro-confluent'
    Network network = Network.newNetwork();

    KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.3.2"))
        .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
        .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
        .withEnv("KAFKA_TRANSACTION_STATE_LOG_NUM_PARTITIONS", "1")
        .withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "500")
        .withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true")
        .withReuse(true)
        .withNetwork(network);
    kafka.start();
    kafkaPort = kafka.getMappedPort(KAFKA_PORT);

    GenericContainer schemaRegistry = new GenericContainer(DockerImageName.parse("confluentinc/cp-schema-registry:7.3.2"))
        .withExposedPorts(8081)
        .withNetwork(kafka.getNetwork())
        .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost")
        .withEnv("SCHEMA_REGISTRY_LISTENERS", "http://0.0.0.0:8081")
        .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "PLAINTEXT://" + kafka.getNetworkAliases().get(0) + ":9092");
    schemaRegistry.start();
    schemaRegistryPort = schemaRegistry.getMappedPort(8081);
  }

  /**
   * Given a resource filename and optional Kafka / Schema Registry ports, return the resource
   * file contents as a String with ports substituted for KAFKA_PORT and SCHEMA_REGISTRY_PORT
   * placeholders.
   *
   * @param resourceFileName    the resource file name
   * @param kafkaPort           the port that Kafka broker exposes
   * @param schemaRegistryPort  the port that Schema Registry exposes
   * @return resource file contents with port values substituted for placeholders
   * @throws IOException if resource file can't be read
   */
  protected static String getResourceFileContents(
      String resourceFileName,
      Optional<Integer> kafkaPort,
      Optional<Integer> schemaRegistryPort
  ) throws IOException {
    URL url = Resources.getResource(resourceFileName);
    String contents = Resources.toString(url, StandardCharsets.UTF_8);
    if (kafkaPort.isPresent()) {
      contents = contents.replaceAll("KAFKA_PORT", kafkaPort.get().toString());
    }
    if (schemaRegistryPort.isPresent()) {
      contents = contents.replaceAll("SCHEMA_REGISTRY_PORT", schemaRegistryPort.get().toString());
    }
    return contents;
  }

  /**
   * Given a resource filename, return the resource file contents as a String.
   *
   * @param resourceFileName    the resource file name
   * @return resource file contents
   * @throws IOException if resource file can't be read
   */
  protected static String getResourceFileContents(
      String resourceFileName
  ) throws IOException {
    // no Kafka / Schema Registry ports
    return getResourceFileContents(resourceFileName, Optional.empty(), Optional.empty());
  }

  /**
   * Utility method to convert a String containing multiple lines into a set of String's where
   * each String is one line. This is useful for creating Flink SQL integration tests based on
   * the tableau results printed via the Table API where the order of results is nondeterministic.
   *
   * @param s multiline String
   * @return set of String's where each member is one line
   */
  protected static Set<String> stringToLineSet(String s) {
    return Sets.newHashSet(Arrays.asList(s.split("\\r?\\n")));
  }

  /**
   * Given a Flink Table API `TableResult` respresenting a SELECT statement result,
   * capture and return the statement's tableau results.
   *
   * @param tableResult Flink Table API `TableResult` respresenting a SELECT statement result
   * @return the SELECT statement's tableau results
   */
  protected static String tableauResults(TableResult tableResult) {
    // capture tableau results printed to stdout in a String
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    System.setOut(new PrintStream(baos));

    // The given table result may come from a table backed by the Kafka or Upsert Kafka connector,
    // both of which perform unbounded (neverending) scans. So, in order to prevent tests from blocking
    // on calls to this method, we kick off a thread to kill the underlying job once output has
    // been printed.
    //
    // Note: as of Flink 1.17.0, the Kafka connector will support bounded scanning, which would obviate
    // the need to do this. However, the Upsert Kafka connector will still be unbounded.
    new Thread(() -> {
      while (0 == baos.size()) {
        try {
          Thread.sleep(500);
        } catch (InterruptedException e) {
          // do nothing; keep waiting
        }
      }
      tableResult.getJobClient().get().cancel();
    }).start();

    try {
      tableResult.print();
    } catch (RuntimeException rte) {
      if (ExceptionUtils.indexOfThrowable(rte, JobCancellationException.class) != -1) {
        // a JobCancellationException in the exception stack is expected due to delayed
        // job cancellation in separate thread; do nothing
      } else {
        rte.printStackTrace();
        System.exit(1);
      }
    }
    System.setOut(System.out);
    return baos.toString();
  }

}

Take a look at this class. It contains the functionality and utility methods that any Flink SQL test would use. Namely, it encapsulates Kafka broker and Schema Registry Testcontainer management and includes utility methods for dynamically configuring Flink SQL Kafka connectors and processing Table API results.

Next, create the test implementation at src/test/java/io/confluent/developer/FlinkSqlAggregatingMinMaxTest.java:

package io.confluent.developer;


import org.apache.flink.table.api.TableResult;
import org.junit.Test;

import java.util.Optional;

import static org.junit.Assert.assertEquals;

public class FlinkSqlAggregatingMinMaxTest extends AbstractFlinkKafkaTest {

  @Test
  public void simpleSelect() throws Exception {
    // create base movie sales table and aggregation table, and populate with test data
    streamTableEnv.executeSql(getResourceFileContents("create-movie-sales.sql.template",
        Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();
    streamTableEnv.executeSql(getResourceFileContents("populate-movie-sales.sql"));
    streamTableEnv.executeSql(getResourceFileContents("create-movie-sales-by-year.sql.template",
        Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();

    // We can't call await() on this result since it won't return. In Flink 17 and later this can change
    // by setting 'scan.bounded.mode' = 'latest-offset' in the movie_sales CREATE TABLE statement, which will
    // cause this INSERT to terminate once the latest offset of movie_sales table is reached.
    streamTableEnv.executeSql(getResourceFileContents("populate-movie-sales-by-year.sql"));

    // execute query on result table that should have movie sales aggregated by release year
    TableResult tableResult = streamTableEnv.executeSql(getResourceFileContents("query-movie-sales-by-year.sql"));

    // Compare actual and expected results. Convert result output to line sets to compare so that order
    // doesn't matter, because the grouped result order doesn't matter -- 2017's could come before or after 2019's.
    String actualTableauResults = tableauResults(tableResult);
    String expectedTableauResults = getResourceFileContents("expected-movie-sales-by-year.txt");
    assertEquals(stringToLineSet(actualTableauResults), stringToLineSet(expectedTableauResults));
  }

}

The test itself is straightforward to follow. It executes the SQL from our resource files, then runs a select statement against the final output TABLE of our application and compares the results to what’s expected.

Invoke the test

Now run the test, which is as simple as:

./gradlew test

Deploy on Confluent Cloud

Run your app with Confluent Cloud

Instead of running a local Kafka cluster, you may use Confluent Cloud, a fully managed Apache Kafka service.

Sign up for Confluent Cloud, a fully managed Apache Kafka service.
After you log in to Confluent Cloud Console, click Environments in the lefthand navigation, click on Add cloud environment, and name the environment learn-kafka. Using a new environment keeps your learning resources separate from your other Confluent Cloud resources.
From the Billing & payment section in the menu, apply the promo code CC100KTS to receive an additional $100 free usage on Confluent Cloud (details). To avoid having to enter a credit card, add an additional promo code CONFLUENTDEV1. With this promo code, you will not have to enter a credit card for 30 days or until your credits run out.
Click on LEARN and follow the instructions to launch a Kafka cluster and enable Schema Registry.

Next, from the Confluent Cloud Console, click on Clients to get the cluster-specific configurations, e.g., Kafka cluster bootstrap servers and credentials, Confluent Cloud Schema Registry and credentials, etc., and set the appropriate parameters in your client application.

Now you’re all set to run your streaming application locally, backed by a Kafka cluster fully managed by Confluent Cloud.

How to find the min/max in a stream of events

Question:

Example use case:

Hands-on code example: