使用 Beam Python SDK 和 PortableRunner 通过 SSL 连接到 Kafka答案

【问题标题】：Using the Beam Python SDK and PortableRunner to connect to Kafka with SSL使用 Beam Python SDK 和 PortableRunner 通过 SSL 连接到 Kafka
【发布时间】：2021-05-06 02:48:44
【问题描述】：

我有下面的代码，用于使用python beam sdk 连接到 kafka。我知道ReadFromKafka 转换是在java sdk 工具（docker 容器）中运行的，但我无法弄清楚如何在sdk 工具的docker 环境中使ssl.truststore.location 和ssl.keystore.location 可访问。 job_endpoint 参数指向 java -jar beam-runners-flink-1.10-job-server-2.27.0.jar --flink-master localhost:8081

pipeline_args.extend([
    '--job_name=paul_test',
    '--runner=PortableRunner',
    '--sdk_location=container',
    '--job_endpoint=localhost:8099',
    '--streaming',
    "--environment_type=DOCKER",
    f"--sdk_harness_container_image_overrides=.*java.*,{my_beam_sdk_docker_image}:{my_beam_docker_tag}",
])

with beam.Pipeline(options=PipelineOptions(pipeline_args)) as pipeline:
    kafka = pipeline | ReadFromKafka(
        consumer_config={
            "bootstrap.servers": "bootstrap-server:17032",
            "security.protocol": "SSL",
            "ssl.truststore.location": "/opt/keys/client.truststore.jks", # how do I make this available to the Java SDK harness 
            "ssl.truststore.password": "password",
            "ssl.keystore.type": "PKCS12",
            "ssl.keystore.location": "/opt/keys/client.keystore.p12", # how do I make this available to the Java SDK harness 
            "ssl.keystore.password": "password",
            "group.id": "group",
            "basic.auth.credentials.source": "USER_INFO",
            "schema.registry.basic.auth.user.info": "user:password"
        },
        topics=["topic"],
        max_num_records=2,
        # expansion_service="localhost:56938"
    )

    kafka | beam.Map(lambda x: print(x))

我尝试将image override 选项指定为--sdk_harness_container_image_overrides='.*java.*,beam_java_sdk:latest' - 其中beam_java_sdk:latest 是我基于apache/beam_java11_sdk:2.27.0 的docker 映像，它在其entrypoint.sh 中提取凭据。但是 Beam 似乎没有使用它，我明白了

INFO  org.apache.beam.runners.fnexecution.environment.DockerEnvironmentFactory  - Still waiting for startup of environment apache/beam_java11_sdk:2.27.0 for worker id 1-1

在日志中。很快就不可避免地紧随其后

Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: Failed to load SSL keystore /opt/keys/client.keystore.p12 of type PKCS12

总之，我的问题是，在 Apache Beam 中，是否可以从 python 梁 sdk 使文件在 java sdk 使用 docker 容器中可用？如果可以，怎么做？

非常感谢。

【问题讨论】：

标签： apache-kafka apache-flink apache-beam apache-beam-kafkaio

【解决方案1】：

目前，没有直接的方法可以实现这一点。正在进行讨论和跟踪问题以提供对此类扩展服务定制的支持（请参阅here、here、BEAM-12538 和BEAM-12539）。这是简短的答案。

答案是肯定的，你可以这样做。您必须将 &paste ExpansionService.java 复制到您的代码库并构建您的自定义扩展服务，您可以在其中指定默认环境（DOCKER）和默认环境配置（您的图像）here。然后您必须手动运行此扩展服务并使用 ReadFromKafka 的expansion_service 参数指定其地址。

【讨论】：