使用docker将kafka主题数据写入redis

Question

我正在使用this repository kafka 连接到redis。

说明：我要做的是使用docker将kafka主题数据写入redis。他们创建了一个自述文件来指导如何设置 kafka 的配置：

curl -s -X POST -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors

connector.json 文件包含：

{
  "config" : {
    "name" : "RedisSinkConnector1",
    "connector.class" : "com.github.jcustenborder.kafka.connect.redis.RedisSinkConnector",
    "tasks.max" : "1",
    "topics" : "mostafa"
  }
}

问题：我知道如何在 kafka 中创建新主题，但问题是我不知道如何更改 docker-compose 或测试连接。虽然我在 kafka 中创建了一个新主题，但 redis 数据库中没有显示任何内容！

如果有人可以帮助我，我将不胜感激。

Answer 1

对于初学者，那里的 compose 文件中没有 Kafka Connect 容器，因此您需要添加一个容器，或者在主机上的 Docker 之外启动 Kafka Connect。

然后，不清楚你是否正确加载了 Redis 连接器，所以打开 http://localhost:8083/connector-plugins 看看是否正确（这也将验证你已经启动了连接服务器}

完成后，您可以发布配置（您需要删除隐藏 curl 输出的 -s）。发布后，您将需要检查正在运行的 Connect 进程的日志，或者您也可以转到 http://localhost:8083/connectors/RedisSinkConnector1/status

鉴于您所展示的内容，并且您已经走到了这一步，以上两者都可能说明了 localhost:6379 的连接异常，因为这是默认连接。您必须提供 "redis.hosts": "redis:6379" 作为属性。

然后，文档中也提到了

此连接器希望来自 Kafka 的记录具有存储为字节或字符串的键和值

因此，将键和值转换器也添加到您的属性以及指定数据类型不会有什么坏处。如果你直接使用 Confluent 中的 Connect 容器，它可能设置为使用 Avro 转换器，而不是字符串或字节一

这是一个您可以发布的有效配置示例

{
  "name" : "RedisSinkConnector1",
  "config" : {
    "connector.class" : "com.github.jcustenborder.kafka.connect.redis.RedisSinkConnector",
    "tasks.max" : "1",
    "topics" : "mostafa",
    "redis.hosts": "redis:6379",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}

通过这些调整，我认为发送任何简单的键值消息都可以，然后使用 redis-cli 运行一些扫描/获取键查询

Answer 2

下面的配置可以解决问题。

{
  "name" : "RedisSinkConnector1",
  "config" : {
    "connector.class" : "com.github.jcustenborder.kafka.connect.redis.RedisSinkConnector",
    "tasks.max" : "1",
    "topics" : "mostafa",
    "redis.hosts": "redis:6379",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}

在 docker-compose 文件中添加 kafka-connect

kafka-connect:
    hostname: kafka-connect
    image: confluentinc/cp-kafka-connect:latest
    container_name: kafka-connect
    ports:
      - 8083:8083
    depends_on:
      - schema-registry
      **- redis**
    environment:
      CONNECT_BOOTSTRAP_SERVERS: kafka:9092
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: "quickstart-avro"
      CONNECT_CONFIG_STORAGE_TOPIC: "quickstart-avro-config"
      CONNECT_OFFSET_STORAGE_TOPIC: "quickstart-avro-offsets"
      CONNECT_STATUS_STORAGE_TOPIC: "quickstart-avro-status"
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      **CONNECT_REST_ADVERTISED_HOST_NAME: "kafka-connect"**
      CONNECT_LOG4J_ROOT_LOGLEVEL: DEBUG
      CONNECT_PLUGIN_PATH: "/usr/share/java,/etc/kafka-connect/jars"
    volumes:
      - $PWD/jars:/etc/kafka-connect/jars

depends_on redis 和 CONNECT_REST_ADVERTISED_HOST_NAME 变量对于解决此问题非常重要