👉目录

1 背景

2 部署 Flink

3 Demo：Topic 复制

4 Demo：单词计数

5 Flink Connector 用法总结

6 注意事项

01

背景

Apache Flink 是一个开源的流处理和批处理框架，具有高吞吐量、低延迟的流式引擎，支持事件时间处理和状态管理，以及确保在机器故障时的容错性和一次性语义。Flink 的核心是一个分布式流数据处理引擎，支持 Java、Scala、Python 和 SQL 编程语言，可以在集群或云环境中执行数据流程序。它提供了 DataStream API 用于处理有界或无界数据流，DataSet API 用于处理有界数据集，以及 Table API 和 SQL 接口用于关系型流和批处理。目前 Flink 最新已经迭代至 1.20 版本，在此过程中不光是 Flink 框架，插件本身也有部分 API 以及配置存在变更，本文主要针对较高版本的 1.17 Flink Pulsar 插件进行测试验证，目前 Flink 版本如下：https://nightlies.apache.org/flink/

02

部署 Flink

2.1 设置 Flink 环境配置

参考 Flink 1.17 官方文档，部署 Flink Docker 版本。

https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/resource-providers/standalone/docker/#getting-started

首先配置 Flink 集群 JobManager 和 TaskManager 环境信息，注意由于 Connector Pulsar 会使用到堆外内存，并且默认任务的堆外内存为 0，因此此处需要显式声明堆外内存大小 taskmanager.memory.task.off-heap.size，这里设置为 1GB。

https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/memory/mem_setup_tm/#configure-off-heap-memory-direct-or-native

    
     $ FLINK_PROPERTIES=$'\njobmanager.rpc.address: jobmanager\ntaskmanager.memory.task.off-heap.size: 1gb\ntaskmanager.memory.process.size: 4gb'$ docker network create flink-network

2.2 部署 JobManager

配置环境变量后部署 JobManager，这里默认映射端口为 8081，部署后登录 8081 端口可以看到 Flink Dashboard 信息。

$ docker run \    --rm \    --name=jobmanager \    --network flink-network \    --publish 8081:8081 \    --env FLINK_PROPERTIES="${FLINK_PROPERTIES}" \    flink:1.17.2-scala_2.12 jobmanager

2.3 部署 TaskManager

JobManager 是维护协调任务的组件，部署 JobManager 后还需要部署具体运行任务的 TaskManager。

$ docker run \    --rm \    --name=taskmanager \    --network flink-network \    --env FLINK_PROPERTIES="${FLINK_PROPERTIES}" \    flink:1.17.2-scala_2.12 taskmanager

运行 TaskManager 后，可以在 8081 JobManager 控制台看到 TaskManager 已经被成功注册，至此 Flink Docker 组件部署完成。

2.4 下载 Flink Cli

在本地编译打包 Pulsar 任务后，还需要使用 Flink Cli 提交本地任务到 Flink Docker 集群，从下方网址下载与当前 Docker 版本一致的 Flink 二进制文件并且解压到本地。

https://flink.apache.org/downloads/

03

Demo：Topic 复制

参考 Flink Pulsar Connector 社区文档和 Oceanus 相关文档，Demo 使用 1.17 版本 Flink SDK 将命名空间的一个 Topic 消息全部复制到另一个 Topic 中，Demo 主要展示 Flink Connector 的基础用法，没有使用自定义序列化器及反序列化器，而是使用的是 Connector 内置的 String 序列化器。

https://cloud.tencent.com/document/product/849/85885#pulsar-source-.E5.92.8C-sink-.E7.A4.BA.E4.BE.8B

https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#apache-pulsar-connector

3.1 主要逻辑

核心逻辑见下方代码，首先使用 ParameterTool 工具解析命令行中传入的参数，之后根据参数信息使用 Connector Source 和 Sink Builder 方法创建一个从 InputTopic 中获取消息发送到 OutputTopic 的 Flink Stream。

public static void main(String[] args) throws Exception {    final ParameterTool parameterTool = ParameterTool.fromArgs(args);    if (parameterTool.getNumberOfParameters() < 2) {        System.err.println("Missing parameters!");        return;    }    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();    env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));    env.enableCheckpointing(60000);    env.getConfig().setGlobalJobParameters(parameterTool);    String brokerServiceUrl = parameterTool.getRequired("broker-service-url");    String inputTopic = parameterTool.getRequired("input-topic");    String outputTopic = parameterTool.getRequired("output-topic");    String subscriptionName = parameterTool.get("subscription-name", "testDuplicate");    String token = parameterTool.getRequired("token");
    //  source    PulsarSource<String> source = PulsarSource.builder()            .setServiceUrl(brokerServiceUrl)            .setStartCursor(StartCursor.latest())            .setTopics(inputTopic)            .setDeserializationSchema(new SimpleStringSchema())            .setSubscriptionName(subscriptionName)            .setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token)            .build();    DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
    //  sink    PulsarSink<String> sink = PulsarSink.builder()            .setServiceUrl(brokerServiceUrl)            .setTopics(outputTopic)            .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)            .setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token)            .setConfig(PulsarSinkOptions.PULSAR_BATCHING_ENABLED, false)            .setSerializationSchema(new SimpleStringSchema())            .build();    stream.sinkTo(sink);    env.execute("Pulsar Streaming Message Duplication");}

3.2 验证

在 TDMQ Pulsar 版控制台创建流入 Topic NinjaDuplicationInput1 和流出 Topic NinjaDuplicationOutput1。

代码编译为 Jar 包后，本地上传 Jar 包到 Flink Docker，从角色管理界面获取具有生产和消费角色的 Token，命令如下所示：

/usr/local/services/flink/flink-1.17.2 # /usr/local/services/flink/flink-1.17.2/bin/flink run /tmp/wordCount/pulsar-flink-examples-0.0.1-SNAPSHOT-jar-with-dependencies.jar \    --broker-service-url http://pulsar-xxxxx \    --input-topic pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaDuplicationInput1 \    --output-topic pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaDuplicationOutput1 \    --subscription-name ninjaTest1 \    --token eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVc
Job has been submitted with JobID c1bdab89c01ef16e00579bd2c6648859

提交任务后，可以看到 Flink Dashboard 出现对应任务，并且状态处于 Running。

在命令行往 NinjaDuplicationInput1 Topic 发送消息。

/usr/local/services/pulsar/apache-pulsar-2.9.5/bin/pulsar-client \--url http://pulsar-xxxxxx \--auth-params token:eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVc  \--auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \produce \-m "i am the bone of my sword" \-n 5 \pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaDuplicationInput1

消息发送完成后，可以在消息查询控制台观察到目标 Topic NinjaDuplicationOutput1 也出现了五条消息，并且消息内容和发送消息一致。

查看 Docker TaskManager 标准输出也能观察到 Sink 往目标 Topic 发送消息的日志。

04

Demo：单词计数

单词计数作为 Flink 中最常见的 Demo，能够比较好的阐述 Flink 的流处理思想。此 Demo 参考 StreamNative 的 Demo，使用 1.17 Flink SDK，将 Pulsar Topic 作为源和目标资源，统计源 Topic 消息中每个时间窗口各个单词出现的次数，并且将结果投递到目标 Topic 中。

https://github.com/streamnative/examples/blob/master/pulsar-flink/README.md

4.1 主要逻辑

整体 Demo 项目文件见下方链接

pulsar-flink-example.zip

file:////tencent/api/attachments/s3/url?attachmentid=20260421

核心逻辑见下方代码，首先使用 ParameterTool 工具解析命令行中传入的参数，之后使用 Flink 内置的反序列化器解析消息体为字符串，在数据处理部分使用系统时间窗口统计时间窗内流入的消息，并且对于每个出现的单词汇聚生成 WordCount 对象，最后使用自定义的序列化器，将 WordCount 对象序列化为 Json 字节数组，投递到目标 Topic 中。

目前 TDMQ Pulsar Connector 支持 Pulsar Schema、Flink Schema 以及自定义序列化器三种方法将 Java 对象序列化为 Pulsar Sink 的字节数组消息体。推荐代码使用自定义序列化器的方式序列化定义的 WordCount 对象。

https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#serializer

还需要注意默认 Sink 配置是开启 Batch Send 模式的，在控制台消息查询时，Batch Message 只会查询到 Batch 中的第一条消息，不利于对照消息数量，Demo 中关闭了 Batch Send 功能。

/** * 参考 streamNative pulsar flink demo * <a href="https://github.com/streamnative/examples/tree/master/pulsar-flink">pulsar-flink example</a> * 由于上方链接的 streamNative flink demo 使用 1.10.1 版本 flink 以及 2.4.17 版本 pulsar connector, * 与当前 1.20 社区版本的 flink 和 pulsar connector api 已经存在部分 api 差异 * 因此本 demo 使用 1.17 flink 版本进行重构 * <a href="https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/resource-providers/standalone/overview/">1.17 flink doc</a> * <p> * demo 统计时间窗口内源 topic 所有消息中每个单词出现频率次数 * 并且将统计结果按照每个单词对应一条消息的格式，序列化后消息后投递到目标 topic 中 * */public class PulsarStreamingWordCount {    private static final Logger LOG = LoggerFactory.getLogger(PulsarStreamingWordCount.class);
    public static void main(String[] args) throws Exception {        //  解析任务传参        //  默认使用 authToken 方式鉴权        final ParameterTool parameterTool = ParameterTool.fromArgs(args);        if (parameterTool.getNumberOfParameters() < 2) {            System.err.println("Missing parameters!");            return;        }        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));        env.enableCheckpointing(60000);        env.getConfig().setGlobalJobParameters(parameterTool);        String brokerServiceUrl = parameterTool.getRequired("broker-service-url");        String inputTopic = parameterTool.getRequired("input-topic");        String outputTopic = parameterTool.getRequired("output-topic");        String subscriptionName = parameterTool.get("subscription-name", "WordCountTest");        String token = parameterTool.getRequired("token");        int timeWindowSecond = parameterTool.getInt("time-window", 60);
        //  source        PulsarSource<String> source = PulsarSource.builder()                .setServiceUrl(brokerServiceUrl)                .setStartCursor(StartCursor.latest())                .setTopics(inputTopic)                //  此处将 message 中的 payload 序列化成字符串类型                //  目前 source 只支持解析消息 payload 中的内容，将 payload 中的内容解析成 pulsar schema 对象或者自定义的 class 对象                //  而无法解析 message 中 properties 中的其他属性，例如 publish_time                //  如果需要解析 message 中的 properties，需要在继承类中实现 PulsarDeserializationSchema.getProducedType() 方法                //  getProducedType 这个方法实现较为繁琐，需要声明每个反序列化后的属性                //  https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#deserializer                .setDeserializationSchema(new SimpleStringSchema())                .setSubscriptionName(subscriptionName)                .setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token)                .build();        //  由于此处没有使用消息体中的时间，即没有使用消息的 publish_time        //  因此此处使用 noWatermark 模式，使用 taskManager 的时间作为时间窗口        DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
        //  process        //  解析 source 中每行消息，通过空格分割成单个单词，之后进行汇聚处理并且初始化成 WordCount 结构体        //  这里使用 TumblingProcessingTimeWindows，即使用当前 taskManager 系统时间计算时间窗口        DataStream<WordCount> wc = stream                .flatMap((FlatMapFunction<String, WordCount>) (line, collector) -> {                    LOG.info("current line = {}, word list = {}", line, line.split("\\s"));                    for (String word : line.split("\\s")) {                        collector.collect(new WordCount(word, 1, null));                    }                })                .returns(WordCount.class)                .keyBy(WordCount::getWord)                .window(TumblingProcessingTimeWindows.of(Time.seconds(timeWindowSecond)))                .reduce((ReduceFunction<WordCount>) (c1, c2) -> {                    WordCount reducedWordCount = new WordCount(c1.getWord(), c1.getCount() + c2.getCount(), null);                    LOG.info("previous [{}] [{}], current wordCount {}", c1, c2, reducedWordCount);                    return reducedWordCount;                });
        //  sink        //  目前 1.17 flink 序列化提供了两种已经实现的方法，一种是使用 pulsar 内置 schema，另一种是使用 flink 的 schema        //  但由于目前 tdmq pulsar 提供的是 2.9 版本的 pulsar，对于 schema 支持还不够完善        //  此处使用 flink PulsarSerializationSchema<T> 提供的接口，当前主要需要实现 serialize(IN element, PulsarSinkContext sinkContext) 方法        //  将传入的 IN 对象自定义序列化为 byte 数组        //  https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#serializer        PulsarSink<WordCount> sink = PulsarSink.builder()                .setServiceUrl(brokerServiceUrl)                .setTopics(outputTopic)                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)                .setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token)                .setConfig(PulsarSinkOptions.PULSAR_BATCHING_ENABLED, false)                .setSerializationSchema(new PulsarSerializationSchema<WordCount>() {                    private ObjectMapper objectMapper;
                    @Override                    public void open(                            SerializationSchema.InitializationContext initializationContext,                            PulsarSinkContext sinkContext,                            SinkConfiguration sinkConfiguration)                            throws Exception {                        objectMapper = new ObjectMapper();                    }
                    @Override                    public PulsarMessage<?> serialize(WordCount wordCount, PulsarSinkContext sinkContext) {                        //  此处将 wordCount 添加处理时间后，将 wordCount 使用 json 方式序列化为 byte 数组                        //  以便能够直接查看消息体内容                        byte[] wordCountBytes;                        wordCount.setSinkDateTime(LocalDateTime.now().toString());                        try {                            wordCountBytes = objectMapper.writeValueAsBytes(wordCount);                        } catch (Exception exception) {                            wordCountBytes = exception.getMessage().getBytes();                        }                        return PulsarMessage.builder(wordCountBytes).build();                    }
                })                .build();        wc.sinkTo(sink);        env.execute("Pulsar Streaming WordCount");    }
}

4.2 验证

在 TDMQ Pulsar 版控制台创建流入 Topic NinjaWordCountInput1 和流出 Topic NinjaWordCountOutput1。

代码编译为 Jar 包后，本地上传 Jar 包到 Flink Docker，从角色管理界面获取具有生产和消费角色的 Token，命令如下所示。

/usr/local/services/flink/flink-1.17.2 # /usr/local/services/flink/flink-1.17.2/bin/flink run /tmp/wordCount/pulsar-flink-examples-0.0.1-SNAPSHOT-jar-with-dependencies.jar \    --broker-service-url http://pulsar-xxxx \    --input-topic pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaWordCountInput1 \    --output-topic pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaWordCountOutput1 \    --subscription-name ninjaTest3 \    --token eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVc  Job has been submitted with JobID 6f608d95506f96c3eac012386f840655

提交任务后，可以看到 Flink Dashboard 出现对应任务，并且状态处于 Running。

在命令行往 NinjaWordCountInput1 Topic 发送消息，此处一共发送两批消息，第一批发送 I am the bone of my sword 5 次，第二批发送 Test1 3 次。

/usr/local/services/pulsar/apache-pulsar-2.9.5/bin/pulsar-client \--url http://pulsar-xxx \--auth-params token:eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVc  \--auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \produce \-m "i am the bone of my sword" \-n 5 \pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaWordCountInput1
/usr/local/services/pulsar/apache-pulsar-2.9.5/bin/pulsar-client \--url http://pulsar-g8akj4eow8z8.sap-8ywks40k.tdmq.ap-gz.qcloud.tencenttdmq.com:8080 \--auth-params token:eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVc  \--auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \produce \-m "test1" \-n 3 \pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaWordCountInput1

消息发送完成后，可以在消息查询控制台观察到目标 Topic NinjaWordCountOutput1 出现了 8 条消息。

每条消息为单词名称，单词出现的次数，单词处理的时间点的 Json 字节数组，下图为 am 单词的消息结构，可以发现出现数量与投递消息数吻合，证明任务运行正常。

查看 TaskManager 可以看出消息体，以及每次解析的消息过程。

05

Flink Connector 用法总结

5.1 版本选择

目前 Flink 插件生产和消费经过调研，在不进行管控改造以及非标操作的情况下，能满足基本的 TDMQ Pulsar 版使用需求。截至现在 Apache Flink 已经发布 1.20 版本，目前推荐使用 Apache Flink 1.15-1.17 对应 Pulsar Connector，不推荐使用 1.15 以下版本，1.18 及以上版本可以参考 1.17 版本使用。

下面介绍 1.15 和 1.17 版本 Pulsar Flink Connector 主要配置。Flink 版本对应的 Flink Connector 依赖可以在 Pulsar Connector Dependencies 处获取。

https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#dependency

各个版本文档链接：https://nightlies.apache.org/flink/

5.2 1.17 Flink Pulsar Connector

5.2.1 代码依赖

Java 项目中引入相关依赖，以 Maven 工程为例，在 pom.xml 添加以下依赖：

<dependency>    <groupId>org.apache.flink</groupId>    <artifactId>flink-connector-base</artifactId>    <version>1.17.2</version></dependency><dependency>    <groupId>org.apache.flink</groupId>    <artifactId>flink-connector-pulsar</artifactId>    <version>4.1.0-1.17</version></dependency><dependency>    <groupId>org.apache.flink</groupId>    <artifactId>flink-streaming-java</artifactId>    <version>1.17.2</version></dependency>

5.2.2 Source 代码示例

PulsarSource<String> source = PulsarSource.builder()        .setServiceUrl(brokerServiceUrl)        .setStartCursor(StartCursor.latest())        .setTopics(inputTopic)        .setDeserializationSchema(new SimpleStringSchema())        .setSubscriptionName(subscriptionName)        .setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token)        .build();

5.2.3 Source 参数说明

Connector Source 全部参数可参考官方文档，下表是常用配置参数：

5.2.4 Sink 代码示例

PulsarSink<String> sink = PulsarSink.builder()       .setServiceUrl(brokerServiceUrl)       .setTopics(outputTopic)       .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)       .setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token)       .setSerializationSchema(new SimpleStringSchema())       .build();

5.2.5 Sink 参数说明

Connector Sink 全部参数可参考官方文档，下表是常用配置参数。

5.3 Flink Pulsar Connector

5.3.1 代码依赖

Java 项目中引入相关依赖，以 Maven 工程为例，在 pom.xml 添加以下依赖：

<dependency>    <groupId>org.apache.flink</groupId>    <artifactId>flink-connector-base</artifactId>    <version>1.15.4</version></dependency><dependency>    <groupId>org.apache.flink</groupId>    <artifactId>flink-connector-pulsar</artifactId>    <version>1.15.4</version></dependency><dependency>    <groupId>org.apache.flink</groupId>    <artifactId>flink-streaming-java</artifactId>    <version>1.15.4</version></dependency>

5.3.2 Source 代码示例

PulsarSource<String> source = PulsarSource.builder()        .setServiceUrl(brokerServiceUrl)        .setAdminUrl(brokerServiceUrl)        .setStartCursor(StartCursor.latest())        .setTopics(inputTopic)        .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))        .setSubscriptionName(subscriptionName)    .setSubscriptionType(SubscriptionType.Exclusive)        .setConfig(PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME, "org.apache.pulsar.client.impl.auth.AuthenticationToken")        .setConfig(PulsarOptions.PULSAR_AUTH_PARAMS, token)        .setConfig(PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE, true)        .build();

5.3.3 Source 参数说明

Connector Source 全部参数可参考官方文档，下表是常用配置参数。

5.3.4 Sink 代码示例

PulsarSink<String> sink = PulsarSink.builder()        .setServiceUrl(brokerServiceUrl)        .setAdminUrl(brokerServiceUrl)        .setTopics(outputTopic)        .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)        .setConfig(PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME, "org.apache.pulsar.client.impl.auth.AuthenticationToken")        .setConfig(PulsarOptions.PULSAR_AUTH_PARAMS, token)        .setSerializationSchema(PulsarSerializationSchema.flinkSchema(new SimpleStringSchema()))        .build();

5.3.5 Sink 参数说明

Connector Sink 全部参数可参考官方文档，下表是常用配置参数。

06

注意事项

由于 Connector Pulsar 会使用到堆外内存，并且默认任务的堆外内存为 0，因此执行 Pulsar Job 需要显式声明堆外内存大小 taskmanager.memory.task.off-heap.size，例如 1gb。
SetSerializationSchema 反序列化提供了两种已经实现的方法，一种是使用 Pulsar 内置 Schema，另一种是使用 Flink 的 Schema。但这两种方法都会造成业务代码与 Schema 耦合。目前建议实现 PulsarSerializationSchema接口，主要需要实现 Serialize(IN Element, PulsarSinkContext SinkContext) 方法，将传入的 IN 对象自定义序列化为 Byte 数组。
目前 Sink 默认开启 Enable_batch 批量投递模式，会将消息打包后投递。如果想要关闭批量投递功能，可以配置 SetConfig(PulsarSinkOptions.PULSAR_BATCHING_ENABLED, False)。
Flink 时间窗口支持两种时间获取方式，一种直接使用任务的系统时间 ProcessTime，另一种是事件自带时间 EventTime。但目前 Source 只支持解析消息 Payload 中的内容，将 Payload 中的内容解析成 Pulsar Schema 对象或者自定义的 Class 对象，而无法解析 Message 中 Properties 中的其他属性，例如消息上传时间 Publish_Time。如果需要解析 Message 中的 Properties，根据文档需要在继承类中实现 PulsarDeserializationSchema.getProducedType() 方法。这个方法实现较为繁琐，需要声明每个反序列化后的属性，因此目前建议直接使用 ProcessTime 作为时间窗口时间。
1.16 及以下版本 Flink Source 的 SetSubscriptionType 方法还保留了 Shared 和 Key_Shared 订阅模式，这两种订阅模式依赖事务 Ack 消息，并且只有当任务 Checkpoint 更新时才会统一提交事务和 Ack。但由于目前 TDMQ Pulsar 没有开放事务功能，因此当前不能同时配置 SetSubscriptionType(SubscriptionType.Shared) 和 SetConfig(PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE, False) 参数。
Oceanus 内置 Pulsar Connector 是基于 StreamNative 版本，适配 Flink 1.13-1.14 版本的 Connector，这两个版本较老，与新版本存在较多 API 不兼容，如果使用 Oceanus 内置版本 Pulsar Connector 与高版本 Flink，可能需要较多代码改造。