使用独立模式，我创建了一个连接器和自定义转换，如下所示：

name=rabbitmq-source
connector.class=com.github.jcustenborder.kafka.connect.rabbitmq.RabbitMQSourceConnector
tasks.max=1
rabbitmq.host=rabbitmq-server
rabbitmq.queue=answers
kafka.topic=net.gutefrage.answers
transforms=extractFields
transforms.extractFields.type=net.gutefrage.connector.transforms.ExtractFields$Value
transforms.extractFields.fields=body,envelope.routingKey
transforms.extractFields.structName=net.gutefrage.events

但是对于分布式连接器，连接restapi的put请求的语法是什么？我在文件里找不到任何例子。
已经尝试了以下几种方法：

cat <<EOF >/tmp/connector
{
  "name": "rabbitmq-source",
  "config": {
    "connector.class": "com.github.jcustenborder.kafka.connect.rabbitmq.RabbitMQSourceConnector",
    "tasks.max": "1",
    "rabbitmq.host": "rabbitmq-server",
    "rabbitmq.queue": "answers",
    "kafka.topic": "net.gutefrage.answers",
    "transforms": "extractFields",
    "transforms.extractFields": {
      "type": "net.gutefrage.connector.transforms.ExtractFields$Value",
      "fields": "body,envelope.routingKey",
      "structName": "net.gutefrage.events"
    }
  }
}
EOF

curl -vs --stderr - -X POST -H "Content-Type: application/json" --data @/tmp/connector "http://localhost:8083/connectors"
rm /tmp/connector

或者这也不起作用：

{
  "name": "rabbitmq-source",
  "config": {
    "connector.class": "com.github.jcustenborder.kafka.connect.rabbitmq.RabbitMQSourceConnector",
    "tasks.max": "1",
    "rabbitmq.host": "rabbitmq-server",
    "rabbitmq.queue": "answers",
    "kafka.topic": "net.gutefrage.answers",
    "transforms": "extractFields",
    "transforms.extractFields.type": "net.gutefrage.connector.transforms.ExtractFields$Value",
    "transforms.extractFields.fields": "body,envelope.routingKey",
    "transforms.extractFields.structName": "net.gutefrage.events"
  }
}

对于最后一个变体，我得到以下错误：

{"error_code":400,"message":"Connector configuration is invalid and contains the following 1 error(s):\nInvalid value class net.gutefrage.connector.transforms.ExtractFields for configuration transforms.extractFields.type: Error getting config definition from Transformation: null\nYou can also find the above list of errors at the endpoint `/{connectorType}/config/validate`"}

请注意，使用properties格式，它可以很好地工作（使用landoops在fast-data-dev中创建新的连接器ui。有趣的是，landoop的ui特性“translate to curl”生成与我的第二个示例相同的json）
更新
为了确保landoop、docker和我的自定义转换没有问题，我使用cop3.3.0中的标准分布式属性在分布式模式下启动了zookeeper、broker、schema registry和kafka connect bin/connect-distributed etc/schema-registry/connect-avro-distributed.properties 哪些日志 [2017-09-13 14:07:52,930] INFO Loading plugin from: /opt/connectors/confluent-oss-gf-assembly-1.0.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:176) [2017-09-13 14:07:53,711] INFO Registered loader: PluginClassLoader{pluginLocation=file:/opt/connectors/confluent-oss-gf-assembly-1.0.jar} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:199) [2017-09-13 14:07:53,711] INFO Added plugin 'com.github.jcustenborder.kafka.connect.rabbitmq.RabbitMQSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:132) [2017-09-13 14:07:53,712] INFO Added plugin 'net.gutefrage.connector.transforms.ExtractFields$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:132) [2017-09-13 14:07:53,712] INFO Added plugin 'net.gutefrage.connector.transforms.ExtractFields$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:132) 目前一切正常。然后我创建了一个连接器配置： cat <<EOF >/tmp/connector { "name": "rabbitmq-source", "config": { "connector.class": "com.github.jcustenborder.kafka.connect.rabbitmq.RabbitMQSourceConnector", "tasks.max": "1", "rabbitmq.host": "rabbitmq-server", "rabbitmq.queue": "answers", "kafka.topic": "net.gutefrage.answers", "transforms": "extractFields", "transforms.extractFields.type": "org.apache.kafka.connect.transforms.ExtractField$Value", "transforms.extractFields.field": "body" } } EOF 请注意，我现在使用的是标准（捆绑）提取字段转换。当我把它贴在 curl -vs --stderr - -X POST -H "Content-Type: application/json" --data @/tmp/connector "http://localhost:8083/connectors" 我也一样 {"error_code":400,"message":"Connector configuration is invalid and contains the following 1 error(s):\nInvalid value class org.apache.kafka.connect.transforms.ExtractField for configuration transforms.extractFields.type: Error getting config definition from Transformation: null\nYou can also find the above list of errors at the endpoint/{connectorType}/config/validate"}*