0

我正在使用 Confluent 的 Kafka 连接器将数据从 Kafka 接收到 MinIO 存储桶。我在io.confluent.connect.s3.S3SinkConnectorKubernetes 环境中使用。这是我当前的 S3 Sink 配置：

kafkaConnectSpec:
  class: io.confluent.connect.s3.S3SinkConnector
  config:
     tasks.max: 1
     topics: topic-name
     s3.bucket.name: bucket
     flush.size: 1
     store.url: http://minio:9000
     format.class: io.confluent.connect.s3.format.json.JsonFormat
     storage.class: io.confluent.connect.s3.storage.S3Storage

云环境部署后，客户希望能够动态控制主题（即随意添加和删除主题）。虽然我明白为什么这可能不理想，但我还是屈服于上级。

因此，为了执行主题添加，我使用了 Kafka REST API：

def update_sink(topic, connector):
 configuration = requests.get("http://kafka-connect-cluster-connect-api:8083/connectors/" + str(connector)).json()

    if "config" not in configuration:
        return {
            "status": 500,
            "message": "Kafka Sink " + str(connector) + " does not have a configuration"
        }

    # Topics must be comma delimited
    if "topics" in configuration["config"]:
        configuration["config"]["topics"] += ',' + topic

    requests.put("http://kafka-connect-cluster-connect-api:8083/connectors/" + str(connector) + "/config", json=configuration["config"])
    print("http://kafka-connect-cluster-connect-api:8083/connectors/" + str(connector) + "/config")
    print(configuration["config"])
    return {
        "status": 200,
        "message": "Kafka Sink " + str(connector) + " successfully updated"
    }

我知道代码并不漂亮，但它现在可以完成工作。/connectors/my-sink/config它本质上向端点发出了一个 PUT 请求，并附加了我的新主题。

这行得通。我的接收器有新主题，我可以发送消息。

但是，在 3-5 分钟内，我的 Kafka Sink Pod 开始重新启动（我认为）Kafka 连接器：

2021-03-19 23:02:55,086 INFO [Worker clientId=connect-1, groupId=connect-cluster] Connector minio-s3-sink config updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [KafkaBasedLog Work Thread - connect-cluster-configs]
2021-03-19 23:02:55,589 INFO [Worker clientId=connect-1, groupId=connect-cluster] Handling connector-only config update by restarting connector minio-s3-sink (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2021-03-19 23:02:55,589 INFO Stopping connector minio-s3-sink (org.apache.kafka.connect.runtime.Worker) [DistributedHerder-connect-1-1]
2021-03-19 23:02:55,589 INFO Shutting down S3 connector minio-s3-sink (io.confluent.connect.s3.S3SinkConnector) [DistributedHerder-connect-1-1]
2021-03-19 23:02:55,598 INFO Stopped connector minio-s3-sink (org.apache.kafka.connect.runtime.Worker) [DistributedHerder-connect-1-1]
2021-03-19 23:02:55,598 INFO [Worker clientId=connect-1, groupId=connect-cluster] Starting connector minio-s3-sink (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
... # Performs the restart here

此时 pod 将失去主题。

我相信这是由config.action.reload = restart配置值引起的。我认为在收到新配置后，连接器将在 N 分钟后重新启动。但是，我似乎找不到任何关于如何改变这种行为的文档。也许我应该在我的 PUT 请求期间这样做，但这感觉很糟糕。这也只是在黑暗中的一个镜头。

有谁知道为什么我的连接器在更新配置的 PUT 请求后重新启动？有没有办法防止这种情况发生？

编辑#1：我试图添加config.action.reload = none，但是连接器仍然重新启动。

我查看了 Kafka Operator 中的日志，并没有触发重置。似乎与 Kafka 连接器操作员完全隔离。

4

1 回答 1

2

该问题记录在 Strimzi 中：

如果启用了 KafkaConnectors，则集群操作员将还原直接使用 Kafka Connect REST API 进行的手动更改

https://strimzi.io/docs/operators/latest/deploying.html#availability_of_the_kafka_connect_rest_api

我不知道这发生在我的部署中，但显然我们必须将其关闭。这是不幸的，因为 K8 连接器部署对于一个简单的开始来说是不错的。

这是将它们“关闭”的相关配置：

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
  name: kafka-connect-cluster
  annotations:
#  # use-connector-resources configures this KafkaConnect
#  # to use KafkaConnector resources to avoid
#  # needing to call the Connect REST API directly
    strimzi.io/use-connector-resources: "false"

这strimzi.io/use-connector-resources: "false"将使您无法通过 YAML 文件添加连接器，但您可以通过 REST API 添加连接器（只要 pod 运行，这些更改就会持续存在）

于 2021-03-24T00:14:43.257 回答