Kafka 连接器使用与开发

Kafka 连接器介绍

Kafka 连接器通常用来构建数据管道，一般有两种使用场景：

开始和结束的端点：例如，将 Kafka 中的数据导出到 HBase 数据库，或者把 Oracle 数据库中的数据导入 Kafka 中。
数据传输的中间介质：例如，为了把海量的日志数据存储到 Elasticsearch 中，可以先把这些日志数据传输到 Kafka 中，然后再从 Kafka 中将这些数据导入到 Elasticsearch 中进行存储。Kafka 连接器可以作为数据管道各个阶段的缓冲区，将消费者程序和生产者程序有效地进行解耦。

Kafka 连接器分为两种：

Source 连接器：负责将数据导入 Kafka。
Sink 连接器：负责将数据从 Kafka 系统中导出。

连接器作为 Kafka 的一部分，是随着 Kafka 系统一起发布的，无须独立安装。

Kafka 连接器特性

Kafka 连接器包含以下特性：

1.是一种处理数据的通用框架，Kafka 连接器指定了一种标准，用来约束 Kafka 与其他系统的集成，简化了 Kafka 连接器的开发、部署和管理过程。
2.提供单机模式和分布式模式：Kafka 连接器支持两种模式，既能扩展到支持大型集群，也可以缩小到开发和测试小规模的集群。
3.提供 REST 接口：使用 REST API 来提交请求并管理 Kafka 连接器。
4.自动管理偏移量：Kafka 连接器可以自动管理偏移量。
5.分布式和可扩展：Kafka 连接器建立在现有的组管理协议上，可以通过添加更多的连接器实例来实现水平扩展，实现分布式服务。
6.数据流和批量集成：利用 Kafka 已有的能力，Kafka 连接器是桥接数据流和批处理系统的一种理想的解决方案。

Kafka 连接器核心概念

连接器实例：连接器实例决定了消息数据的流向，即消息从何处复制，以及将复制的消息写入到何处。连接器实例负责 Kafka 与其他系统之间的逻辑处理，连接器实例通常以 JAR 包形式存在，通过实现 Kafka 系统应用接口来完成。
任务数：在分布式模式下，每一个连接器实例可以将一个作业切分成多个任务（Task），然后再将任务分发到各个事件线程（Worker）中去执行。任务不会保存当前的状态信息，通常由特定的 Kafka Topic 来保存，例如，指定具体属性 offset.storage.topic 和 status.storage.topic 的值来保存。
事件线程：在 Kafka 中，连接器实例和任务数都是逻辑层面的，需要由具体的线程来执行，事件线程包含两种模式--单机模式和分布式模式。
转换器：转换器能将字节数据转换成 Kafka 连接器的内部格式，也能将 Kafka 连接器内部存储的数据格式转换成字节数据。

使用 Kafka 连接器

单机模式

单机模式配置文件

配置单机模式连接器相关参数 config/connect-standalone.properties：

# Kafka 集群 broker 地址
bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
# 指定键值对 JSON 转换器类
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# 启用键值对转换器
key.converter.schemas.enable=true
value.converter.schemas.enable=true
# 单机模式设置偏移量存储文件
offset.storage.file.filename=/tmp/connect.offsets
# 设置偏移量持久化时间间隔
offset.flush.interval.ms=10000

将数据从文件导入 Kafka Topic 中

编辑 Kafka 连接器配置文件 config/connect-file-source.properties：

# 设置连接器名字
name=local-file-source
# 指定连接器类
connector.class=FileStreamSource
# 设置最大任务数
tasks.max=1
# 指定读取的文件
file=/tmp/test.txt
# 指定写入 Kafka 的 Topic 
topic=connect_test

创建数据源文件并添加数据：

[root@kafka1 ~]# cat /tmp/test.txt
kafka
hadoop
kafka-connect

启动一个单机模式的连接器将数据导入 Kafka Topic 中：

[root@kafka1 kafka]# connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties

启动消费者程序查看导入到 connect_test 主题中的数据：

[root@kafka1 ~]# kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic connect_test -from-beginning
{"schema":{"type":"string","optional":false},"payload":"kafka"}
{"schema":{"type":"string","optional":false},"payload":"hadoop"}
{"schema":{"type":"string","optional":false},"payload":"kafka-connect"}
{"schema":

当往文件中追加数据时，消费者可以消费到新的数据：

[root@kafka1 ~]# echo java >> /tmp/test.txt 
[root@kafka1 ~]# echo python  >> /tmp/test.txt

消费者消费到的新的数据：

{"type":"string","optional":false},"payload":"java"}
{"schema":{"type":"string","optional":false},"payload":"python"}

将 Kafka Topic 中的数据导出到文件

编辑 Kafka 连接器配置文件 config/connect-file-sink.properties：

# 设置连接器名字
name=local-file-sink
# 指定连接器类
connector.class=FileStreamSink
# 设置最大任务数
tasks.max=1
# 将数据写入的文件
file=/tmp/sink.txt
# 指定导出数据的 Kafka 的 Topic
topics=connect_test

启动一个单机模式的连接器将 Kafka Topic 中的数据导出：

[root@kafka1 kafka]# connect-standalone.sh  config/connect-standalone.properties config/connect-file-sink.properties

查看导出文件的内容：

[root@kafka1 ~]# cat /tmp/sink.txt 
python
kafka
hadoop
kafka-connect
java

分布式模式

在分布式模式下， Kafka 连接器会自动均衡每个事件线程所处理的任务数。允许用户动态地增加或者减少任务，在执行任务、修改配置、提交偏移量时能够得到容错保障。

在分布式模式下，Kafka 连接器会在 Kafka Topic 中存储偏移量，配置和任务状态（单机模式下是保持在本地文件中）。建议手动创建存储偏移量的主题，这样可以按需设置主题的分区数和副本数。

在分布式模式下， Kafka 连接器的配置文件不能使用命令行，需要使用 REST API 来执行创建，修改和销毁 Kafka 连机器的操作。

创建连接器相关主题

# 创建偏移量的的存储主题
kafka-topics.sh --create --bootstrap-server kafka1:9092 --replication-factor 3 --partitions 1 --topic connect-offsets
# 创建配置存储主题
kafka-topics.sh --create --bootstrap-server kafka1:9092 --replication-factor 3 --partitions 6 --topic connect-configs
# 创建任务状态存储主题
kafka-topics.sh --create --bootstrap-server kafka1:9092 --replication-factor 3 --partitions 6 --topic connect-status

分布式模式配置文件

# 设置 Kafka 集群地址
bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
# 设置连接器唯一组名称
group.id=connect-cluster
# 指定键值对 JSON 转换器类
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# 启用键值对转换器
key.converter.schemas.enable=true
value.converter.schemas.enable=true
# 设置偏移量的的存储主题
offset.storage.topic=connect-offsets
# 设置配置存储主题
config.storage.topic=connect-configs
# 设置任务状态存储主题
status.storage.topic=connect-status
# 设置偏移量持久化时间间隔
offset.flush.interval.ms=10000

启动分布式模式连接器：

[root@kafka1 kafka]# connect-distributed.sh config/connect-distributed.properties

查看连接器版本号信息：

[root@kafka1 ~]# curl http://kafka1:8083
{"version":"2.7.0","commit":"448719dc99a19793","kafka_cluster_id":"wp8iI172SaqLHqNvEh3T-w"}

查看当前已安装的插件：

[root@kafka1 ~]# curl http://kafka1:8083/connector-plugins -s | jq
[
  {
    "class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "type": "sink",
    "version": "2.7.0"
  },
  {
    "class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "type": "source",
    "version": "2.7.0"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
    "type": "source",
    "version": "1"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
    "type": "source",
    "version": "1"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "type": "source",
    "version": "1"
  }
]

连接器 API 接口

由于 Kafka 连接器旨在作为服务运行，因此还提供了用于管理连接器的 REST API。默认情况下，此服务在端口 8083 上运行。以下是当前支持的 API 接口：

GET /connectors #返回活动连接器的列表
POST /connectors #创建一个新的连接器; 请求主体应该是包含字符串name字段和config带有连接器配置参数的对象字段的JSON对象
GET /connectors/{name} #获取有关特定连接器的信息
GET /connectors/{name}/config #获取特定连接器的配置参数
PUT /connectors/{name}/config #更新特定连接器的配置参数
GET /connectors/{name}/status #获取连接器的当前状态，包括连接器是否正在运行，失败，已暂停等，分配给哪个工作者，失败时的错误信息以及所有任务的状态
GET /connectors/{name}/tasks #获取当前为连接器运行的任务列表
GET /connectors/{name}/tasks/{taskid}/status #获取任务的当前状态，包括如果正在运行，失败，暂停等，分配给哪个工作人员，如果失败，则返回错误信息
PUT /connectors/{name}/pause #暂停连接器及其任务，停止消息处理，直到连接器恢复
PUT /connectors/{name}/resume #恢复暂停的连接器（或者，如果连接器未暂停，则不执行任何操作）
POST /connectors/{name}/restart #重新启动连接器（通常是因为失败）
POST /connectors/{name}/tasks/{taskId}/restart #重启个别任务（通常是因为失败）
DELETE /connectors/{name} #删除连接器，停止所有任务并删除其配置
#Kafka Connect还提供了用于获取有关连接器插件信息的REST API：
GET /connector-plugins #返回安装在Kafka Connect集群中的连接器插件列表。请注意，API仅检查处理请求的worker的连接器，这意味着您可能会看到不一致的结果，尤其是在滚动升级期间，如果添加新的连接器jar
PUT /connector-plugins/{connector-type}/config/validate # 根据配置定义验证提供的配置值。此API执行每个配置验证，在验证期间返回建议值和错误消息。

将数据从文件导入到 Kafka Topic 中

通过 REST API 请求创建一个新的连接器实例，将数据导入到 Kafka Topic 中。这里使用的是 Chrome 浏览器上名为 API Tester 的插件：

{
    "name": "distributed-console-source", #自定义连接器名字
    "config":
    {
        "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
        "tasks.max": "1",
        "topic": "distributed_connect_test", #创建的topic
        "file": "/tmp/distributed_test.txt"  #读取的文件
    }
}

查看刚刚创建的连接器：

[root@kafka1 ~]# curl http://kafka1:8083/connectors -s | jq
[
  "distributed-console-source"
]

此时开启一个消费者实例可以成功消费到 Kafka Topic 中的数据：

[root@kafka1 ~]# kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic distributed_connect_test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"distributed_kafka"}
{"schema":{"type":"string","optional":false},"payload":"kafka"}
{"schema":{"type":"string","optional":false},"payload":"hadoop"}

通过 REST API 请求创建一个新的连接器实例，将数据从 Kafka Topic 中导出到文件中。

{
    "name": "distributed-console-sink",
    "config":
    {
        "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
        "tasks.max": "1",
        "topics": "distributed_connect_test", #Kafka 中存在的 Topic
        "file": "/tmp/distributed_sink.txt" #导出数据到指定文件
    }
}

查看目前的连接器：

[root@kafka1 ~]# curl http://kafka1:8083/connectors -s | jq
[
  "distributed-console-sink",
  "distributed-console-source"
]

查看文件可以看到数据成功从 Kafka Topic 中导出：

[root@kafka1 ~]# cat /tmp/distributed_sink.txt 
distributed_kafka
kafka
hadoop

开发 Kafka 连接器插件

开发一个完整的 Kafka 连接器插件，分为两部分来实现：

编写 Source 连接器。Source 连接器负责将第三方系统的数据导入 Kafka Topic 中。
编写 Sink 连接器。Sink 连接器负责将 Kafka Topic 中的数据导出到第三方系统中。

第三方系统可以是关系型数据库（如 MySQL、Oracle 等）、文件系统（如本地文件，分布式文件系统等）、日志系统等。

本实例使用的是 Maven 工程，需要在 pom.xml 文件中引入 Kafka 依赖包:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>0.10.2.1</version>
</dependency>

编写 Source 连接器

编写一个自定义的 Source 连接器，需要实现两个抽象类：

SourceConnector 类，用来初始化连接器配置和任务数。
SourceTask 类，用来实现标准输入或者文件读取。

编写输入连接器实例

package book_8;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.source.SourceConnector;
/**
 * 输入连接器，用来实现读取配置信息和分配任务等一些初始化工作
 * @author 程治玮
 * @since 2021/3/16 9:51 下午
 */
public class CustomerFileStreamSourceConnector extends SourceConnector {
    // 定义主题配置变量
    public static final String TOPIC_CONFIG = "topic";
    // 定义文件配置变量
    public static final String FILE_CONFIG = "file";
    // 实例化一个配置对象
    private static final ConfigDef CONFIG_DEF = new ConfigDef().define(FILE_CONFIG, Type.STRING, Importance.HIGH, "Source filename.").define(TOPIC_CONFIG, Type.STRING, Importance.HIGH, "The topic to publish data to");
    // 声明文件名变量
    private String filename;
    // 声明主题变量
    private String topic;
    /** 获取版本. */
    public String version() {
        return AppInfoParser.getVersion();
    }
    /** 开始初始化. */
    public void start(Map<String, String> props) {
        filename = props.get(FILE_CONFIG);
        topic = props.get(TOPIC_CONFIG);
        if (topic == null || topic.isEmpty())
            throw new ConnectException("FileStreamSourceConnector configuration must include 'topic' setting");
        if (topic.contains(","))
            throw new ConnectException("FileStreamSourceConnector should only have a single topic when used as a source.");
    }
    /** 实例化输入类. */
    public Class<? extends Task> taskClass() {
        return CustomerFileStreamSourceTask.class;
    }
    /** 获取配置信息. */
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        ArrayList<Map<String, String>> configs = new ArrayList<>();
        Map<String, String> config = new HashMap<>();
        if (filename != null)
            config.put(FILE_CONFIG, filename);
        config.put(TOPIC_CONFIG, topic);
        configs.add(config);
        return configs;
    }
    @Override
    public void stop() {
    }
    /** 获取配置对象. */
    public ConfigDef config() {
        return CONFIG_DEF;
    }
}

编写输入连接器任务类

package book_8;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 输入连接器任务类，用来实现标准输入或者文件读取
 * @author 程治玮
 * @since 2021/3/16 9:47 下午
 */
public class CustomerFileStreamSourceTask extends SourceTask {
    // 声明一个日志类
    private static final Logger LOG = LoggerFactory.getLogger(CustomerFileStreamSourceTask.class);
    // 定义文件字段
    public static final String FILENAME_FIELD = "filename";
    // 定义偏移量字段
    public static final String POSITION_FIELD = "position";
    // 定义值的值的数据格式
    private static final Schema VALUE_SCHEMA = Schema.STRING_SCHEMA;
    // 声明文件名
    private String filename;
    // 声明输入流对象
    private InputStream stream;
    // 声明读取对象
    private BufferedReader reader = null;
    // 定义缓冲区大小
    private char[] buffer = new char[1024];
    // 声明偏移量变量
    private int offset = 0;
    // 声明主题名
    private String topic = null;
    // 声明输入流偏移量
    private Long streamOffset;
    /** 获取版本. */
    public String version() {
        return new CustomerFileStreamSourceConnector().version();
    }
    /** 开始执行任务. */
    public void start(Map<String, String> props) {
        filename = props.get(CustomerFileStreamSourceConnector.FILE_CONFIG);
        if (filename == null || filename.isEmpty()) {
            stream = System.in;
            streamOffset = null;
            reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8));
        }
        topic = props.get(CustomerFileStreamSourceConnector.TOPIC_CONFIG);
        if (topic == null)
            throw new ConnectException("FileStreamSourceTask config missing topic setting");
    }
    /** 读取记录并返回数据集. */
    public List<SourceRecord> poll() throws InterruptedException {
        if (stream == null) {
            try {
                stream = new FileInputStream(filename);
                Map<String, Object> offset = context.offsetStorageReader().offset(Collections.singletonMap(FILENAME_FIELD, filename));
                if (offset != null) {
                    Object lastRecordedOffset = offset.get(POSITION_FIELD);
                    if (lastRecordedOffset != null && !(lastRecordedOffset instanceof Long))
                        throw new ConnectException("Offset position is the incorrect type");
                    if (lastRecordedOffset != null) {
                        LOG.debug("Found previous offset, trying to skip to file offset {}", lastRecordedOffset);
                        long skipLeft = (Long) lastRecordedOffset;
                        while (skipLeft > 0) {
                            try {
                                long skipped = stream.skip(skipLeft);
                                skipLeft -= skipped;
                            } catch (IOException e) {
                                LOG.error("Error while trying to seek to previous offset in file: ", e);
                                throw new ConnectException(e);
                            }
                        }
                        LOG.debug("Skipped to offset {}", lastRecordedOffset);
                    }
                    streamOffset = (lastRecordedOffset != null) ? (Long) lastRecordedOffset : 0L;
                } else {
                    streamOffset = 0L;
                }
                reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8));
                LOG.debug("Opened {} for reading", logFilename());
            } catch (FileNotFoundException e) {
                LOG.warn("Couldn't find file {} for FileStreamSourceTask, sleeping to wait for it to be created", logFilename());
                synchronized (this) {
                    this.wait(1000);
                }
                return null;
            }
        }
        try {
            final BufferedReader readerCopy;
            synchronized (this) {
                readerCopy = reader;
            }
            if (readerCopy == null)
                return null;
            ArrayList<SourceRecord> records = null;
            int nread = 0;
            while (readerCopy.ready()) {
                nread = readerCopy.read(buffer, offset, buffer.length - offset);
                LOG.trace("Read {} bytes from {}", nread, logFilename());
                if (nread > 0) {
                    offset += nread;
                    if (offset == buffer.length) {
                        char[] newbuf = new char[buffer.length * 2];
                        System.arraycopy(buffer, 0, newbuf, 0, buffer.length);
                        buffer = newbuf;
                    }
                    String line;
                    do {
                        line = extractLine();
                        if (line != null) {
                            LOG.trace("Read a line from {}", logFilename());
                            if (records == null)
                                records = new ArrayList<>();
                            records.add(new SourceRecord(offsetKey(filename), offsetValue(streamOffset), topic, null, null, null, VALUE_SCHEMA, line, System.currentTimeMillis()));
                        }
                    } while (line != null);
                }
            }
            if (nread <= 0)
                synchronized (this) {
                    this.wait(1000);
                }
            return records;
        } catch (IOException e) {
        }
        return null;
    }
    /** 解析一条记录. */
    private String extractLine() {
        int until = -1, newStart = -1;
        for (int i = 0; i < offset; i++) {
            if (buffer[i] == '\n') {
                until = i;
                newStart = i + 1;
                break;
            } else if (buffer[i] == '\r') {
                if (i + 1 >= offset)
                    return null;
                until = i;
                newStart = (buffer[i + 1] == '\n') ? i + 2 : i + 1;
                break;
            }
        }
        if (until != -1) {
            String result = new String(buffer, 0, until);
            System.arraycopy(buffer, newStart, buffer, 0, buffer.length - newStart);
            offset = offset - newStart;
            if (streamOffset != null)
                streamOffset += newStart;
            return result;
        } else {
            return null;
        }
    }
    /** 停止任务. */
    public void stop() {
        LOG.trace("Stopping");
        synchronized (this) {
            try {
                if (stream != null && stream != System.in) {
                    stream.close();
                    LOG.trace("Closed input stream");
                }
            } catch (IOException e) {
                LOG.error("Failed to close FileStreamSourceTask stream: ", e);
            }
            this.notify();
        }
    }
    private Map<String, String> offsetKey(String filename) {
        return Collections.singletonMap(FILENAME_FIELD, filename);
    }
    private Map<String, Long> offsetValue(Long pos) {
        return Collections.singletonMap(POSITION_FIELD, pos);
    }
    /** 判断是标准输入还是读取文件. */
    private String logFilename() {
        return filename == null ? "stdin" : filename;
    }
}

编写 Sink 连接器

在 Kafka 系统中，实现一个自定义的 Sink 连接器，需要实现两个抽象类。

SinkTask 类：用来实现标准输出或者文件写入。
SinkConnector 类：用来初始化连接器配置和任务数。

编写输出连接器实例

package book_8;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.sink.SinkConnector;
/**
 * 输出连接器，用来实现读取配置信息和分配任务等一些初始化工作
 * @author 程治玮
 * @since 2021/3/16 9:56 下午
 */
public class CustomerFileStreamSinkConnector extends SinkConnector {
    // 声明文件配置变量
    public static final String FILE_CONFIG = "file";
    // 实例化一个配置对象
    private static final ConfigDef CONFIG_DEF = new ConfigDef().define(FILE_CONFIG, Type.STRING, Importance.HIGH, "Destination filename.");
    // 声明一个文件名变量
    private String filename;
    /** 获取版本信息. */
    public String version() {
        return AppInfoParser.getVersion();
    }
    /** 执行初始化. */
    public void start(Map<String, String> props) {
        filename = props.get(FILE_CONFIG);
    }
    /** 实例化输出类.*/
    public Class<? extends Task> taskClass() {
        return CustomerFileStreamSinkTask.class;
    }
    /** 获取配置信息. */
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        ArrayList<Map<String, String>> configs = new ArrayList<>();
        for (int i = 0; i < maxTasks; i++) {
            Map<String, String> config = new HashMap<>();
            if (filename != null)
                config.put(FILE_CONFIG, filename);
            configs.add(config);
        }
        return configs;
    }
    public void stop() {
    }
    /** 获取配置对象. */
    public ConfigDef config() {
        return CONFIG_DEF;
    }
}

编写输出连接器任务类

package book_8;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Map;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 输出连接器任务类，用来实现标准输出或者文件写入
 * @author 程治玮
 * @since 2021/3/16 9:54 下午
 */
public class CustomerFileStreamSinkTask extends SinkTask {
    // 声明一个日志对象
    private static final Logger LOG = LoggerFactory.getLogger(CustomerFileStreamSinkTask.class);
    // 声明一个文件名变量
    private String filename;
    // 声明一个输出流对象
    private PrintStream outputStream;
    /** 构造函数. */
    public CustomerFileStreamSinkTask() {
    }
    /** 重载构造函数. */
    public CustomerFileStreamSinkTask(PrintStream outputStream) {
        filename = null;
        this.outputStream = outputStream;
    }
    /** 获取版本号. */
    public String version() {
        return new CustomerFileStreamSinkConnector().version();
    }
    /** 开始执行任务. */
    public void start(Map<String, String> props) {
        filename = props.get(CustomerFileStreamSinkConnector.FILE_CONFIG);
        if (filename == null) {
            outputStream = System.out;
        } else {
            try {
                outputStream = new PrintStream(new FileOutputStream(filename, true), false, StandardCharsets.UTF_8.name());
            } catch (FileNotFoundException | UnsupportedEncodingException e) {
                throw new ConnectException("Couldn't find or create file for FileStreamSinkTask", e);
            }
        }
    }
    /** 发送记录给Sink并输出. */
    public void put(Collection<SinkRecord> sinkRecords) {
        for (SinkRecord record : sinkRecords) {
            LOG.trace("Writing line to {}: {}", logFilename(), record.value());
            outputStream.println(record.value());
        }
    }
    /** 持久化数据. */
    public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
        LOG.trace("Flushing output stream for {}", logFilename());
        outputStream.flush();
    }
    /** 停止任务. */
    public void stop() {
        if (outputStream != null && outputStream != System.out)
            outputStream.close();
    }
    /** 判断是标准输出还是文件写入. */
    private String logFilename() {
        return filename == null ? "stdout" : filename;
    }
}

打包与部署

将编写好的连接器代码打成 JAR 包，放在每台 Kafka 的 libs目录下，然后重启 Kafka 集群和分布式模式连接器。

启动完成后，可以通过下面命令查看已安装的连接器插件，可以看到两个自定义开发的连接器插件已经部署成功：

[root@kafka1 ~]# curl http://kafka1:8083/connector-plugins -s | jq
[
  # 自定义的 Sink 连接器插件
  {
    "class": "book_8.CustomerFileStreamSinkConnector",
    "type": "sink",
    "version": "2.7.0"
  },
 # 自定义的 Source 连接器插件
  {
    "class": "book_8.CustomerFileStreamSourceConnector",
    "type": "source",
    "version": "2.7.0"
  },
  {
    "class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "type": "sink",
    "version": "2.7.0"
  },
  {
    "class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "type": "source",
    "version": "2.7.0"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
    "type": "source",
    "version": "1"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
    "type": "source",
    "version": "1"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "type": "source",
    "version": "1"
  }
]