Kafka producer自动故障切换/故障回复

Question

我想知道是否可以在Kafka制作程序中配置2个不同的Kafka集群。

目前我正试图让我的制片人

我正在使用Apache Kafka 2.8和Python 3.7的confluent_kafka==1.8.2包。

生产商代码下方：

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

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

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

其他选择是，

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