我正在开发python应用程序,它连接到kafka并使用一些数据。你知道吗
def main(argv):
params = parse_arg(argv)
logging.info("Connecting to topic\t" + params.tasks_topic)
consumer = KafkaConsumer(params.tasks_topic,
group_id='kafkatester',
bootstrap_servers=params.kafka.split(','),
auto_offset_reset='latest',
enable_auto_commit=False,
max_poll_records=1,
max_poll_interval_ms=18000)
def parse_arg(argv):
parser = argparse.ArgumentParser()
parser.add_argument('-k', '--kafka')
parser.add_argument('-t', '--tasks-topic')
args = parser.parse_args()
return AppParams(args.kafka, args.tasks_topic)
在当地一切正常。但是,当我在docker中运行它时,得到了意想不到的结果:
08/09/2019 09:54:57 AM Connecting to topic taskstest
08/09/2019 09:54:57 AM <BrokerConnection node_id=bootstrap-0 host=MySecretIP:9092 <connecting> [IPv4 ('MySecretIP', 9092)]>: connecting to MySecretIP:9092 [('MySecretIP', 9092) IPv4]
08/09/2019 09:54:57 AM Probing node bootstrap-0 broker version
08/09/2019 09:54:57 AM <BrokerConnection node_id=bootstrap-0 host=MySecretIP:9092 <connecting> [IPv4 ('MySecretIP', 9092)]>: Connection complete.
08/09/2019 09:54:57 AM Broker version identifed as 1.0.0
08/09/2019 09:54:57 AM Set configuration api_version=(1, 0, 0) to skip auto check_version requests on startup
Traceback (most recent call last):
File "/usr/local/lib/python3.7/runpy.py", line 193, in _run_module_as_main
"__main__", mod_spec)
File "/usr/local/lib/python3.7/runpy.py", line 85, in _run_code
exec(code, run_globals)
File "/app/src/main.py", line 40, in <module>
main(sys.argv[1:])
File "/app/src/main.py", line 26, in main
max_poll_interval_ms=18000)
File "/usr/local/lib/python3.7/site-packages/kafka/consumer/group.py", line 390, in __init__
self._subscription.subscribe(topics=topics)
File "/usr/local/lib/python3.7/site-packages/kafka/consumer/subscription_state.py", line 120, in subscribe
self.change_subscription(topics)
File "/usr/local/lib/python3.7/site-packages/kafka/consumer/subscription_state.py", line 169, in change_subscription
self._ensure_valid_topic_name(t)
File "/usr/local/lib/python3.7/site-packages/kafka/consumer/subscription_state.py", line 142, in _ensure_valid_topic_name
raise ValueError('Topic name "{0}" is illegal, it contains a character other than ASCII alphanumerics, ".", "_" and "-"'.format(topic))
" is illegal, it contains a character other than ASCII alphanumerics, ".", "_" and "-"
看起来KafkaConsumer
无法处理params.tasks_topic
变量。。为什么?你知道吗
kafka的python版本是1.4.6
,python版本是3.7.3
。你知道吗
你需要删减这个话题 看起来您发送的主题名称带有空格或空格 所以你需要首先验证主题是str.Trimmed str 所以我的建议是这样的:
我已经测试过了,它就像一个魔咒;) 如果需要帮助,请打电话给我;)
相关问题 更多 >
编程相关推荐