3

我正在尝试使用 robinhood / faust 但没有成功！

我已经成功地在我的 confluent-kafka localhost 实例中创建了一个插入原始主题的生产者！

但浮士德无法连接到本地主机。

我的应用程序.py：

import faust
import base64
import random
from datetime import datetime


SOURCE_TOPIC="input_msgs"
TARGET_TOPIC="output_msgs"

app = faust.App("messages-stream", 
    broker="kafka://"+'localhost:9092',
    topic_partitions=1,
    store="memory://")

class OriginalMessage(faust.Record):
    msg: str


class TransformedMessage(faust.Record):
    msg_id: int
    msg_data: str
    msg_base64: str
    created_at: float 
    source_topic: str
    target_topic: str
    deleted: bool

topic = app.topic(SOURCE_TOPIC, value_type=OriginalMessage)
out_topic = app.topic(TARGET_TOPIC, partitions=1)

table = app.Table(
    "output_msgs",
    default=TransformedMessage,
    partitions=1,
    changelog_topic=out_topic,
)

print('Initializing Thread Processor...')


@app.agent(topic)
async def transformedmessage(messageevents):
    async for transfmessage in messageevents:
        try:

            table[transfmessage.msg_id] = random.randint(1, 999999)
            table[transfmessage.msg_data] = transfmessage.msg
            table[transfmessage.msg_base64] = base64.b64encode(transfmessage.msg)
            table[transfmessage.created_at] = datetime.now().isoformat()
            table[transfmessage.source_topic] = SOURCE_TOPIC
            table[transfmessage.target_topic] = TARGET_TOPIC
            table[transfmessage.deleted] = False

        except Exception as e:
            print(f"Error: {e}")


if __name__ == "__main__":
    app.main()

错误

[2020-01-24 18:05:36,910] [55712] [ERROR] Unable connect to node with id 1: [Errno 8] nodename nor servname provided, or not known 
[2020-01-24 18:05:36,910] [55712] [ERROR] [^Worker]: Error: ConnectionError('No connection to node with id 1') 

    "No connection to node with id {}".format(node_id))
kafka.errors.ConnectionError: ConnectionError: No connection to node with id 1

我正在运行：faust -A app worker -l debug

4

1 回答 1

0

我遇到了这个错误，幸运的是我有点预料到它，所以找出问题并不难。

在 Confluent 中，您必须配置应该用于访问所有为您引导的 Kafka 代理的域。我真的不知道该域有多重要，所以我只是随机添加一些东西，直到我被卡住。

当然，我像你一样被困在这里，所以我启动了 Wireshark 来看看 Faust 和引导服务器之间发生了什么。事实证明，引导对话是这样的：

..............faust-1.10.4..PLAIN <-- client name
................PLAIN             <-- authentication protocol
....foobar.foobar.foobar2000.     <-- credentials!
b0.svs.cluster.local..#.......b1.svs.cluster.local..# <-- individual Kafka brokers

这些遵循我选择的域名模式，并在 Confluent 文档中进行了描述：https ://docs.confluent.io/current/installation/operator/co-endpoints.html

如果这些名称没有解析，您会在此处得到模糊的错误，因为尽管引导成功，Kafka 客户端因此无法实际连接到端点。因此，选择一个您实际控制的域，或者将您需要的答案放在本地/etc/hosts或等效文件中。

17  123.111.222.1 b0.svs.cluster.local
18  123.111.222.2 b1.svs.cluster.local

重启 Faust 后，bootstrap 和 Kafka 连接成功。

于 2020-05-04T13:54:33.207 回答