Kafka 生产者自动故障转移/故障恢复

Question

我想知道是否可以在 Kafka 生产者中配置 2 个不同的 Kafka 集群。

目前我正在尝试让我的生产者和消费者自动故障回复到被动集群，而无需重新配置 (bootstrap.servers) 并重新启动他们的应用程序。

我将 Apache Kafka 2.8 和 confluent_kafka==1.8.2 包与 Python 3.7 一起使用。

生产者代码下方：

from time import sleep

from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'clusterA:32531, clusterB:30804'})


def delivery_report(err, msg):
    """ Called once for each message produced to indicate delivery result.
        Triggered by poll() or flush(). """
    if err is not None:
        print('Message delivery failed: {}'.format(err))
    else:

        print(f'Message delivered to {msg.offset()}')


with open('test_data.csv', 'r') as read_obj:
    csv_reader = reader(read_obj)
    header = next(csv_reader)
    # Check file as empty
    if header is not None:
        # Iterate over each row after the header in the csv
        for row in csv_reader:
            sleep(0.02)
            p.produce(topic='demo', key=row[5], value=str(row), callback=delivery_report)
p.flush()

当我杀死 clusterB 时，我收到以下错误消息。

%4|1643837239.074|CLUSTERID|rdkafka#producer-1| [thrd:main]: Broker clusterA:32531/bootstrap reports different ClusterId "MLWCRsVXSxOf2YGPRIivjA" than previously known "6ZtcQCRPQ5msgeD3r7I11w": a client must not be simultaneously connected to multiple clusters
%3|1643837240.995|FAIL|rdkafka#producer-1| [thrd:clusterB:30804/bootstrap]: 172.27.176.222:30804/bootstrap: Connect to ipv4#clusterB:30804 failed: Unknown error (after 2044ms in state CONNECT)

Answer 1

目前，您必须手动将引导信息更新到辅助集群，这将需要重新启动客户端才能进行故障转移。

以编程方式，为了连接到单独的集群，您必须停止当前的生产者实例并使用新的引导服务器配置启动一个新实例。然而，这可能会变得相当复杂。

其他选项是，

• 您使用 LB 或 VIP 配置 kafka（不推荐，因为本质上需要从客户端直接连接到代理）
• 配置存储引导服务器配置的共享存储（memcached 或 redis）。您的客户端将在引导过程中获取引导服务器。在失败期间，您更改存储中的值并重新启动客户端。 （这使得操作非常简单）