在 Confluent Kafka Python Client 中指定复制因子

Question

我已经设置了 Kafka 的单一代理实例以及 Zookeeper、Kafka-tools、Schema-registry 和 control-center.The 设置是使用 docker compose 和使用 Confluent 提供的 images.Here 是 docker-compose 的样子：

    zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 2181:2181

  broker:
    image: confluentinc/cp-server:latest
    depends_on:
      - zookeeper
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092
      CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper:2181
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: "true"
      CONFLUENT_SUPPORT_CUSTOMER_ID: "anonymous"

  kafka-tools:
    
    image: confluentinc/cp-kafka:latest
    hostname: kafka-tools
    container_name: kafka-tools
    command: ["tail", "-f", "/dev/null"]
    network_mode: "host"

  schema-registry:
    image: confluentinc/cp-schema-registry:5.5.0
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - zookeeper
      - broker
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: "zookeeper:2181"

  control-center:
    image: confluentinc/cp-enterprise-control-center:latest
    hostname: control-center
    container_name: control-center
    depends_on:
      - zookeeper
      - broker
      - schema-registry
    ports:
      - "9021:9021"
    environment:
      CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
      CONTROL_CENTER_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONTROL_CENTER_REPLICATION_FACTOR: 1
      CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
      CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
      CONFLUENT_METRICS_TOPIC_REPLICATION: 1
      PORT: 9021

尝试使用 confluent-kafka python 客户端创建生产者和消费者 application.The 以下是我的 Kafka 生产者的代码：

from datetime import datetime
import os
import json
from uuid import uuid4
from confluent_kafka import SerializingProducer
from confluent_kafka.serialization import StringSerializer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.json_schema import JSONSerializer

class BotExecutionProducer(object):
    """
    Class represents the 
    Bot execution Stats
    """

    def __init__(self,ticketId,accountId,executionTime,status):

        self.ticketId = ticketId
        self.accountId = accountId
        self.executionTime = executionTime
        self.timestamp = str(datetime.now())
        self.status = status


    def botexecution_to_dict(self,botexecution,ctx):
        """
        Returns a Dict representation of the 
        KafkaBotExecution instance
        botexecution : KafkaBotExecution instance
        ctx: SerializaionContext

        """
        return dict(ticketId=self.ticketId,
                    accountId=self.accountId,
                    executionTime=self.executionTime,
                    timestamp=self.timestamp,
                    status=self.status
            )

    def delivery_report(self,err, msg):
        """
        Reports the failure or success of a message delivery.
        """
        if err is not None:
            print("Delivery failed for User record {}: {}".format(msg.key(), err))
            return
        print('User record {} successfully produced to {} [{}] at offset {}'.format(
            msg.key(), msg.topic(), msg.partition(), msg.offset()))

    def send(self):
        """
        Will connect to Kafka Broker
        validate and send the message
        """
        topic = "bots.execution"
        schema_str = """
            {
            "$schema": "http://json-schema.org/draft-07/schema#",
            "title": "BotExecutions",
            "description": "BotExecution Stats",
            "type": "object",
            "properties": {
                "ticketId": {
                "description": "Ticket ID",
                "type": "string"
                },
                "accountId": {
                "description": "Customer's AccountID",
                "type": "string"             
                },
                "executionTime": {
                "description": "Bot Execution time in seconds",
                "type": "number"
                },
                "timestamp": {
                "description": "Timestamp",
                "type": "string"
                },
                "status": {
                "description": "Execution Status",
                "type": "string"
                }
            },
            "required": [ "ticketId", "accountId", "executionTime", "timestamp", "status"]
            }
            """

        schema_registry_conf = {'url': 'http://localhost:8081'}

        schema_registry_client = SchemaRegistryClient(schema_registry_conf)

        json_serializer = JSONSerializer(schema_str,schema_registry_client,self.botexecution_to_dict)

        producer_conf = {
                    'bootstrap.servers': "localhost:9092",
                    'key.serializer': StringSerializer('utf_8'),
                    'value.serializer': json_serializer,
                    'acks': 0,
    }


        producer = SerializingProducer(producer_conf)
        print(f'Producing records to topic {topic}')

        producer.poll(0.0)

        try:
            print(self)
            producer.produce(topic=topic,key=str(uuid4()),partition=1,
                             value=self,on_delivery=self.delivery_report)

        except ValueError:
            print("Invalid Input,discarding record.....")

现在，当我执行代码时，它应该创建一个 Kafka 主题并将 JSON 数据推送到该主题，但这似乎不起作用，它一直显示复制因子为 3 的错误当只有一个时指定 broker.is 有一种方法可以在上面 code.The 中定义复制因子 kafka borker 在使用 Kafka cli 执行相同操作时可以完美工作。有什么我想念的吗？

Answer 1

我不确定 cp-server 和 cp-kafka 图像之间的全部区别，但您可以为自动创建主题的默认复制因子添加一个变量

KAFKA_DEFAULT_REPLICATION_FACTOR: 1

如果这不起作用，请导入并使用 AdminClient

在 Confluent Kafka Python Client 中指定复制因子

Specifying replication factor in Confluent Kafka Python Client

apache-kafka

confluent-platform

confluent-kafka-python