如何配置KSQLDB实现Schema Registry故障自动切换？

问题 #

测试Kafka Schema Registry故障切换场景时遇到异常：已部署主备两个Schema Registry Docker容器，KSQLDB服务器指向主节点时可正常验证消息Schema，但关闭主节点后，KSQLDB无法自动切换到备用节点，导致无法接收Kafka主题数据。当前使用的docker-compose.yml配置如下：

schema-registry:
    image: confluentinc/cp-schema-registry:${CP_VERSION}
    depends_on:
      - zookeeper
      - kafka
    ports:
      - "8081:8081"
    container_name: schema-registry
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092
      SCHEMA_REGISTRY_ACCESS_CONTROL_ALLOW_ORIGIN: '*'
      SCHEMA_REGISTRY_ACCESS_CONTROL_ALLOW_METHODS: 'GET,POST,PUT,OPTIONS'
      SCHEMA_REGISTRY_LEADER_ELIGIBILITY : "true"
      SCHEMA_REGISTRY_GROUP_ID : "schema-registry-group"
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
      
  schema-registry-2:
    image: confluentinc/cp-schema-registry:${CP_VERSION}
    depends_on:
      - kafka
      - schema-registry
    ports:
      - "8082:8082"
    container_name: schema-registry-2
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry-2
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092
      SCHEMA_REGISTRY_ACCESS_CONTROL_ALLOW_ORIGIN: '*'
      SCHEMA_REGISTRY_ACCESS_CONTROL_ALLOW_METHODS: 'GET,POST,PUT,OPTIONS'
      SCHEMA_REGISTRY_LEADER_ELIGIBILITY : "true"
      SCHEMA_REGISTRY_GROUP_ID : "schema-registry-group"
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8082

  primary-ksqldb-server:
    image: ${KSQL_IMAGE_BASE}confluentinc/ksqldb-server:${KSQL_VERSION}
    hostname: primary-ksqldb-server
    container_name: primary-ksqldb-server
    depends_on:
      - kafka
      - schema-registry
    ports:
      - "8088:8088"
    environment:
      KSQL_CONFIG_DIR: "/etc/ksql"
      KSQL_LISTENERS: http://0.0.0.0:8088
      KSQL_BOOTSTRAP_SERVERS: kafka:9092
      KSQL_KSQL_ADVERTISED_LISTENER : http://localhost:8088
      KSQL_KSQL_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"
      KSQL_KSQL_EXTENSION_DIR: "/usr/ksqldb/ext/"
      KSQL_KSQL_SERVICE_ID: "nrt_"
      KSQL_KSQL_STREAMS_NUM_STANDBY_REPLICAS: 1
      KSQL_KSQL_QUERY_PULL_ENABLE_STANDBY_READS: "true"
      KSQL_KSQL_HEARTBEAT_ENABLE: "true"
      KSQL_KSQL_LAG_REPORTING_ENABLE : "true"
      KSQL_KSQL_QUERY_PULL_MAX_ALLOWED_OFFSET_LAG : 100
      KSQL_LOG4J_APPENDER_KAFKA_APPENDER: "org.apache.kafka.log4jappender.KafkaLog4jAppender"
      KSQL_LOG4J_APPENDER_KAFKA_APPENDER_LAYOUT: "io.confluent.common.logging.log4j.StructuredJsonLayout"
      KSQL_LOG4J_APPENDER_KAFKA_APPENDER_BROKERLIST: localhost:9092
      KSQL_LOG4J_APPENDER_KAFKA_APPENDER_TOPIC: KSQL_LOG
      KSQL_LOG4J_LOGGER_IO_CONFLUENT_KSQL: INFO,kafka_appender
      KSQL_KSQL_QUERY_PULL_METRICS_ENABLED: "true"
      KSQL_JMX_OPTS: >
          -Djava.rmi.server.hostname=localhost
          -Dcom.sun.management.jmxremote
          -Dcom.sun.management.jmxremote.port=1099
          -Dcom.sun.management.jmxremote.authenticate=false
          -Dcom.sun.management.jmxremote.ssl=false
          -Dcom.sun.management.jmxremote.rmi.port=1099

解决方案 #

要实现KSQLDB在Schema Registry主节点宕机时自动切换到备用节点，需从集群配置和客户端配置两方面调整：

1. 完善Schema Registry集群配置 #

当前两个节点已设置相同的SCHEMA_REGISTRY_GROUP_ID和SCHEMA_REGISTRY_LEADER_ELIGIBILITY: "true"，这部分符合集群要求，需补充一个关键配置：

• 为两个Schema Registry节点添加SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL: PLAINTEXT（与Kafka的通信协议保持一致）

2. 修改KSQLDB的Schema Registry连接配置 #

当前KSQLDB仅配置了单个Schema Registry地址，需改为多地址逗号分隔，同时添加重试容错参数：
在primary-ksqldb-server的环境变量中修改：

# 替换原单个地址为所有可用节点地址
KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081,http://schema-registry-2:8082"
# 添加重试与故障转移配置
KSQL_KSQL_SCHEMA_REGISTRY_CLIENT_MAX_RETRIES: "5"
KSQL_KSQL_SCHEMA_REGISTRY_CLIENT_RETRY_BACKOFF_MS: "1000"
KSQL_KSQL_SCHEMA_REGISTRY_CLIENT_REQUEST_TIMEOUT_MS: "5000"

各参数作用：

• KSQL_KSQL_SCHEMA_REGISTRY_URL：指定所有可用的Schema Registry节点，客户端会自动尝试故障转移
• MAX_RETRIES/RETRY_BACKOFF_MS：配置重试次数和间隔，避免主节点宕机后立即返回失败
• REQUEST_TIMEOUT_MS：设置请求超时时间，确保客户端能及时切换到备用节点

3. 验证配置生效 #

修改配置后重启所有容器：

docker-compose down && docker-compose up -d

测试步骤：

• 启动所有服务，确认KSQLDB能正常连接Schema Registry并处理数据
• 停止主Schema Registry容器：docker stop schema-registry
• 观察KSQLDB日志，确认其自动切换到备用节点，且能继续接收、验证主题数据

内容的提问来源于stack exchange，提问作者vka