如何在没有连接器的情况下创建合适的Kafka连接插件？

Question

我尝试为我的数据创建带有“转换”的插件到kafka-connect，并将其与不同的接收器连接器一起使用。当我安装插件时，kafka-connect看不到我的类。

我使用kafka connect maven插件创建了我的捆绑包zip。使用confluent hub（来自本地文件）的安装已成功。

所有文件都已解压，我的工作者属性已更新插件。路径。我在分布式模式下运行connect，并尝试从包中创建带有transformer的新连接器。

我的插件结构如下所示：

- mwojdowski-my-connect-plugin-0.0.1-SNAPSHOT
|- manifest.json
|- lib
||- my-connect-plugin-0.0.1-SNAPSHOT.jar

还有我的舱单。json文件：

{
  "name" : "my-connect-plugin",
  "version" : "0.0.1-SNAPSHOT",
  "title" : "my-connect-plugin",
  "description" : "A set of transformations for Kafka Connect",
  "owner" : {
    "username" : "mwojdowski",
    "name" : "Marcin Wojdowski<[e-mail]>"
  },
  "tags" : [ "transform", "field", "topic" ],
  "features" : {
    "supported_encodings" : [ "any" ],
    "single_message_transforms" : true,
    "confluent_control_center_integration" : true,
    "kafka_connect_api" : true
  },
  "documentation_url" : "",
  "docker_image" : { },
  "license" : [ {
    "name" : "Confluent Software License",
    "url" : "https://www.confluent.io/software-evaluation-license"
  } ],
  "component_types" : [ "transform" ],
  "release_date" : "2019-08-29"
}

接下来，我尝试创建新的连接器：

curl -XPOST -H 'Content-type:application/json' 'localhost:8083/connectors' -d '{
    "name" : "custom-file-sink-with-validation",
    "config" : {
    "connector.class" : "FileStreamSink",
        "tasks.max" : "1",
        "topics" : "test_topic",
        "file" : "/tmp/my-plugin-test.txt",
        "key.ignore" : "true",
        "schema.ignore" : "true",
        "drop.invalid.message": "false",
        "behavior.on.malformed.documents": "warn",
        "key.converter":"org.apache.kafka.connect.storage.StringConverter",
        "value.converter":"org.apache.kafka.connect.storage.StringConverter",
        "transforms" : "Validation",
        "transforms.Validation.type" : "org.kafka.connect.my.connector.ValidateId"
    }
}'

重新启动kafka connect后，当我尝试创建新连接器时，会引发异常：

{
    "error_code": 400,
    "message": "Connector configuration is invalid and contains the following 2 error(s):\nInvalid value org.kafka.connect.my.connector.ValidateId for configuration transforms.Validation.type: Class org.kafka.connect.my.connector.ValidateId could not be found.\nInvalid value null for configuration transforms.Validation.type: Not a Transformation\nYou can also find the above list of errors at the endpoint `/{connectorType}/config/validate`"
}

我也尝试手动安装插件，如下文档：https://docs.confluent.io/current/connect/managing/install.html

但看起来好像Connect没有装我的罐子。

当我将jar复制到share/java/kafka时，它可以工作，但这不是一个解决方案。

我怀疑我的插件是skip，因为它不包含连接器。在这种情况下，我应该手动将JAR添加到类路径吗？（与相对）https://docs.confluent.io/current/connect/userguide.html#installing-插件）

或者我应该明确指出我的连接器配置来尝试使用我的插件吗？

你好，M。

Answer 1

对不起，问题真的很琐碎。在重构过程中，其中一个包在最后得到“s”，我错过了在配置中更新它。

"transforms.Validation.type" : "org.kafka.connect.my.connectors.ValidateId"

而不是

"transforms.Validation.type" : "org.kafka.connect.my.connector.ValidateId"

我在从独立切换到分布式之前对其进行了重构。再次抱歉让您担心，并感谢您的支持。

问候你，马辛