Flink Pulsar Integration Related Examples

We provided a set of examples on using the latest flink-connector-pulsar in the Flink repository. Showing the user how to use this connector.

The example list

1. SimpleSource: Consuming the message from Pulsar by using Flink's StringSchema.

How to use

Prepare the environment.

Download and set up a standalone Pulsar locally. Read this tutorial and config the Pulsar one by one.

If you use macOS and Homebrew, you can use our pre-written Formula. Just execute these commands below.

brew tap streamnative/streamnative

## Without install OpenJDK
brew install pulsar

## With an OpenJDK provided by Homebrew
brew install pulsar --with-openjdk

Install pulsarctl

After install and setup the Pulsar standalone, we need some management tools for operating on the Pulsar cluster. We prefer to use pulsarctl because it supports shell auto-completion. You can skip this section if you want to use the scripts bundled in Pulsar distribution.

Mac operation system

Use homebrew to install pulsarctl on the Mac operation system.

brew tap streamnative/streamnative
brew install pulsarctl

We would auto install zsh-completion and bash-completion when you use Homebrew.

Linux operation system

Use this command to install pulsarctl on the Linux operation system.

sh -c "$(curl -fsSL https://raw.githubusercontent.com/streamnative/pulsarctl/master/install.sh)"

Windows operation system

To install pulsarctl on the Windows operation system, follow these steps:

1. Download the package from here.
2. Add the pulsarctl directory to your system PATH.
3. Execute pulsarctl -h to verify that pulsarctl is work.

Prepare the test dataset

All the code snippet shown below was using pulsarctl. You can convert it to Pulsar scripts by reading the documentation for the Pulsar admin interface.

1. Create the sample tenant on Pulsar.

## Mark sure you have the standalone cluster.
pulsarctl clusters list standalone

## Create tenant
pulsarctl tenants create sample --allowed-clusters="standalone"

2. Create the flink namespace below sample tenant.

pulsarctl namespaces create sample/flink

3. Create topic simple-string with 8 partition under flink namespace.

pulsarctl topics create sample/flink/simple-string 8

Execute pulsarctl topics list sample/flink make sure we would list a set of topics like below.

+-----------------------------------------------------+---------------+
|                     TOPIC NAME                      | PARTITIONED ? |
+-----------------------------------------------------+---------------+
| persistent://sample/flink/simple-string             | Y             |
| persistent://sample/flink/simple-string-partition-0 | N             |
| persistent://sample/flink/simple-string-partition-1 | N             |
| persistent://sample/flink/simple-string-partition-2 | N             |
| persistent://sample/flink/simple-string-partition-3 | N             |
| persistent://sample/flink/simple-string-partition-4 | N             |
| persistent://sample/flink/simple-string-partition-5 | N             |
| persistent://sample/flink/simple-string-partition-6 | N             |
| persistent://sample/flink/simple-string-partition-7 | N             |
+-----------------------------------------------------+---------------+

Execute the program in IntelliJ IDEA

We share the IDEA run configuration in .run directory. You can choose the example case in IDEA's Run Configuration and execute it.