【问题标题】:How to deploy the kafka connect in a distributed mode?如何以分布式模式部署kafka connect?
【发布时间】:2020-01-02 02:34:30
【问题描述】:

我正在使用 kubernetes 中的 JDBC sink 连接器构建 Kafka-connect 应用程序。我尝试了独立模式,它正在工作。我想转向分布式模式。 通过运行下面的 yaml 文件,我可以成功构建两个 pod(kafka 连接器):

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  namespace: vtq
  name: kafka-sink-postgres-dis
spec:
  replicas: 2
  template:
    metadata:
      labels:
        app: kafka-sink-postgres-dis
    spec:
      containers:
      - name: kafka-sink-postgres-dis
        image: ***
        imagePullPolicy: Always

bin/connect-distributed.sh config/worker.properties

bootstrap.servers=***:9092
offset.flush.interval.ms=10000

rest.port=8083
rest.host.name=127.0.0.1


key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081


# Prevent the connector from pulling all historical messages
auto.offset.reset=latest

# options below may be required for distributed mode

# unique name for the cluster, used in forming the Connect cluster group. Note that this must not conflict with consumer group IDs
group.id=connect-postgres-sink-dis

# Topic to use for storing offsets. This topic should have many partitions and be replicated.
offset.storage.topic=postgres-connect-offsets
offset.storage.replication.factor=3

# Topic to use for storing connector and task configurations; note that this should be a single partition, highly replicated topic.
# You MUST manually create the topic to ensure single partition for the config topic as auto created topics may have multiple partitions.

config.storage.topic=postgres-connect-configs
config.storage.replication.factor=3

# Topic to use for storing statuses. This topic can have multiple partitions and should be replicated.
status.storage.topic=postgres-connect-status
status.storage.replication.factor=3

并在每个 pod 内创建了一个带有 task.max=1 的接收器连接器,这两个连接器监听了同一个主题。原来他们只是重复了。

curl -X POST -H "Content-Type: application/json" --data '{"name": "postgres_sink_dis", "config": {"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max":"1", "connection.url":"***","topics":"***"}}' http://127.0.0.1:8083/connectors

但我对 kafka 连接集群、工作程序、连接器和任务的概念感到非常困惑。我从 https://github.com/enfuse/kafka-connect-demo/blob/master/docs/install-connector.md。 他们在配置连接器之前设置端口转发到其余端口。我试过了,在部署服务并创建连接器后, curl -s 172.0.0.1:8083/connectors 没有返回。

谁能给我一个简短的描述我下一步应该做什么,任何相关信息都会非常有帮助。谢谢!

更新: 最后,我弄清楚了问题并解决了问题。 1. 分别部署两个 pods/workers,group.id 相同,rest.port 不同(https://docs.confluent.io/current/connect/userguide.html)。 2. 在 pod 内,创建一个带有任务的连接器。

【问题讨论】:

    标签: distributed apache-kafka-connect


    【解决方案1】:

    例如,您有一个由两个工人/两个 pod 组成的连接器集群。您可以在集群中创建具有多个任务的连接器(接收器或源),这些任务将分布在两个工作器中。

    【讨论】:

    • @ Dipperman 我通过将副本设置为 2 将我的 pod 部署在两个节点中。并在每个 pod 内创建了一个接收器连接器,其中 task.max=1 并且这两个连接器监听了相同的主题。原来他们只是重复了。他们向我的 postgres 发送相同的消息。工作并没有真正分发。需要指出的一件事,我的主题设置是默认的,只有一个分区。那会是个问题吗?还是我的配置有问题?
    • 您必须在每个 pod 中配置相同的 group.id。配置了匹配 group.id 值的分布式工作人员会自动发现彼此并形成一个集群。
    • 您应该拥有一个具有副本 2 的部署。如果您使用副本 2 创建两个节点,那么您将以相同的形式创建两个不同的集群,每个集群都有两个工作人员。这就是您在 postgres 中有重复消息的原因。你的 pods/workers 之间应该有 rest.advertised.host.name 不同。
    • 如果不是生产环境:1.-停止连接器,2.-删除主题config.storage.topic,offset.storage.topic,status.storage.topic,3.-重新创建主题 4.-创建一个具有副本两个的部署,每个工作人员具有不同的 rest.advertised.host.name 和每个工作人员具有相同的 group.id 5.-检查是否一切正常 5.-您将拥有一个具有两个工作人员的集群,一个接收器连接器和任何工作人员中的一项任务。
    • 谢谢!在第 4 步,我不知道如何使用副本 2 来获得不同的属性。相反,我分别创建了两个具有相同 group.id 但不同 rest.advertised.host.name 的 pod。当我尝试使用命令 "curl -X POST -H "Content-Type: application/json" --data '{...}' localhost:8083/connectors" 在 pod 内创建连接器时出现问题,错误消息 "{" error_code":500,"message":"无效的 URI 主机:null (authority: sink_conn_w2:8083)"}"。 sink_conn_w2:8083 是 pod 的 leaderURL,sink_conn_w2 是第一个 pod 中的 rest.advertised.host.name。
    猜你喜欢
    • 1970-01-01
    • 2018-08-13
    • 2019-02-03
    • 2019-11-30
    • 2019-05-12
    • 2020-05-17
    • 1970-01-01
    • 2020-08-03
    • 2017-05-27
    相关资源
    最近更新 更多