java - 实现 kafka 连接自定义分区器

Question

我正在使用 confluent 的 kafka 连接将数据通过管道传输到 s3 存储桶中。理想情况下基于键进行分区。由于现有的 FieldPartitioner 仅适用于 Avro 模式记录，而不适用于一般的字符串化 JSON 文本。我以为我会编写自己的连接器。

这是类(class):

package com.package.kafka.connect;


import com.google.gson.JsonElement;
import com.google.gson.JsonParser;

import io.confluent.connect.storage.partitioner.DefaultPartitioner;
import io.confluent.connect.storage.partitioner.FieldPartitioner;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.connect.sink.SinkRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

public class JsonFieldPartitioner<T> extends DefaultPartitioner<T> {

    private static final Logger log = LoggerFactory.getLogger(FieldPartitioner.class);
    private List<String> fieldNames;
    private List<String> keys;

    public void configure(Map<String, Object> config){
        fieldNames = (List<String>) config.get("partition.field.name");
        String field =  fieldNames.get(0);
        keys = new ArrayList<String>(Arrays.asList(field.split(".")));
    }

    public String encodePartition(SinkRecord sinkRecord){
        String value = sinkRecord.value().toString();

        JsonElement rootElement = new JsonParser().parse(value);
        JsonElement element = rootElement;

        for(String key : keys){
            log.info("key: "+ key);
            try{
                element = element.getAsJsonObject().get(key);
            }catch(Exception e){
                log.error("encountered error getting key: " + key);
                throw new ConfigException("Key element not found" + key);
            }
        }

        String fieldValue  = "";
        try{
            fieldValue =  element.getAsString();
        }catch(Exception e){
            log.error("encountered error getting key value ");
            throw new ConfigException("Key element not found");
        }
        return fieldValue;
    }

    public List<T> partitionFields() {
        if (partitionFields == null) {
            partitionFields = newSchemaGenerator(config).newPartitionFields(
                    Utils.join(fieldNames, ",")
            );
        }
        return partitionFields;
    }
}

当我构建它并尝试运行 kafka connect 时出现错误

java.lang.NullPointerException
    at io.confluent.connect.storage.partitioner.PartitionerConfig.classNameEquals(PartitionerConfig.java:269)
    at io.confluent.connect.storage.partitioner.PartitionerConfig.access$000(PartitionerConfig.java:32)
    at io.confluent.connect.storage.partitioner.PartitionerConfig$PartitionerClassDependentsRecommender.visible(

从看 Packaging a custom Java `partitioner.class` plugin for Kafka Connect in Confluent 4.1 + Kafka 1.1?
我尝试将由此构建的 jar 文件放入 kafka-connect-storage-common $CONFLUENT_HOME 中的目录，但我仍然遇到相同的错误。

构建 jar 的 gradle 文件在这里

    id 'java'
}

group 'JsonFieldPartitioner'
version '1.0-SNAPSHOT'

sourceCompatibility = 1.8

repositories {
    mavenCentral()
}

dependencies {
    compile group: 'org.apache.kafka', name: 'connect-api', version: '2.3.0'
    compile fileTree( dir:'/Users/myuser/confluent-5.3.0/share/java/kafka-connect-storage-common', include: ['*.jar'])
    compile group: 'joda-time', name: 'joda-time', version: '2.10.3'
    compile group: 'com.google.code.gson', name: 'gson', version: '2.8.5'
    testCompile group: 'junit', name: 'junit', version: '4.12'
}

在 s3 连接器属性文件中，我只是通过 com.package.kafka.connect.JsonFieldPartitioner 引用了该类

如果有人成功构建自定义分区器，将不胜感激。

Answer 1

将您的自定义 jar 文件复制到插件文件夹下
例子
/usr/local/share/kafka/plugins/confluentinc-kafka-connect-s3/