Kafka Connect 无法将自定义存储接收器分区器转换为分区器接口

Question

我需要为 kafka connect S3 sink plugin 创建一个自定义分区器。 我已经使用 kotlin 在自定义类中扩展了 HourlyPartitioner：

class RawDumpHourlyPartitioner<T> : HourlyPartitioner<T>() {
...
}

并相应地更改了我的连接器配置以使用自定义类：

"partitioner.class": "co.myapp.RawDumpHourlyPartitioner",

然后我创建了我们的 jar（我们使用shadow）并将其包含到基于 kafka 连接镜像的自定义 docker 镜像中（镜像版本与我们在项目中使用的依赖项相同）：

FROM gradle:6.0-jdk8 as builder
WORKDIR /app
ADD . .
RUN gradle clean shadowJar

FROM confluentinc/cp-kafka-connect:5.3.2

COPY --from=builder /app/build/libs/kafka-processor-0.1-all.jar /usr/share/java/kafka/kafka-processor.jar

连接器启动时出现此错误：

ERROR WorkerSinkTask{id=staging-raw-dump-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
java.lang.ClassCastException: co.myapp.RawDumpHourlyPartitioner cannot be cast to io.confluent.connect.storage.partitioner.Partitioner

为了仔细检查，我创建了一个尝试实例化类的 java 文件，它没有抛出任何错误：

import io.confluent.connect.storage.partitioner.Partitioner;

public class InstantiateTest {
    public static void main(String[] args) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
        Class<? extends Partitioner<?>> partitionerClass =
                (Class<? extends Partitioner<?>>) Class.forName("co.myapp.RawDumpHourlyPartitioner");

        Partitioner<?> partitioner = partitionerClass.newInstance();
    }
}

Answer 1

查看 kafka 连接 guide 它说：

Kafka Connect 插件只是一组 JAR 文件，Kafka Connect 可以在其中找到一个或多个连接器、转换和/或转换器的实现。 Kafka Connect 将每个插件相互隔离，以便一个插件中的库不受任何其他插件中的库的影响。这在混合和匹配来自多个提供商的连接器时非常重要。

这意味着，由于我使用的是 S3 接收器连接器，因此我必须将带有自定义分区器的 jar 放在 S3 插件的目录中。

将 jar 文件移动到 /usr/share/java/kafka-connect-s3 解决了这个问题

在我提到的 cmets 中，我的 jar 还包含我们在主 kafka-connect 配置（环境变量）中使用的自定义主题名称策略，在这种情况下，jar 需要位于 /usr/share/java/kafka 文件夹中

更新：正如cricket_007 所说，最好将自定义分区器jar 放入所有其他分区器所在的/usr/share/java/kafka-connect-storage-common 文件夹中

Answer 2

根据您使用的 Sink，我们需要将分区器类推到那里，就像我们使用 Confluent Kafka 5.5 和连接器类 Azure Gen2 Storage 时一样。

为此，我们需要编写类似于以下Repo in Github 的自定义分区器。

然后我们将自定义 JAR 放在以下路径中：

/usr/share/confluent-hub-components/confluentinc-kafka-connect-azure-data-lake-gen2-storage/lib/ 

之后我们的连接器类就成功了！