apache-flink - Flink 到 Nifi 的 Magic Header 不存在

Question

我正在尝试使用this example将 Nifi 连接到 Flink:

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
            .url("http://localhost:8090/nifi")
            .portName("Data for Flink")
            .requestBatchCount(5)
            .buildConfig();

    SourceFunction<NiFiDataPacket> nifiSource = new NiFiSource(clientConfig);
    DataStream<NiFiDataPacket> streamSource = env.addSource(nifiSource).setParallelism(2);

    DataStream<String> dataStream = streamSource.map(new MapFunction<NiFiDataPacket, String>() {
        @Override
        public String map(NiFiDataPacket value) throws Exception {
            return new String(value.getContent(), Charset.defaultCharset());
        }
    });

    dataStream.print();
    env.execute();

我将 Nifi 作为具有默认属性的独立服务器运行，但以下属性除外:

nifi.remote.input.host=localhost
nifi.remote.input.secure=false
nifi.remote.input.socket.port=8090
nifi.remote.input.http.enabled=true

每次调用都会失败，Nifi中的日志如下:

[Site-to-Site Worker Thread-24] o.a.nifi.remote.SocketRemoteSiteListener 
Unable to communicate with remote instance null due to
org.apache.nifi.remote.exception.HandshakeException: Handshake 
with nifi://localhost:61680 failed because the Magic Header 
was not present; closing connection

Nifi版本:1.7.1，Flink版本:1.7.1

Answer 1

使用nifi-toolkit后，我删除了nifi.remote.input.socket.port的自定义值，然后添加了transportProtocol(SiteToSiteTransportProtocol.HTTP) 到我的 SiteToSiteClientConfig 和 http://localhost:8080/nifi 作为 URL。

我首先更改端口的原因是，如果不指定协议(protocol)HTTP，它将默认使用RAW。 当从 Flink 端使用 RAW 协议(protocol)时，客户端无法创建 Transaction 并打印以下警告:

Unable to refresh Remote Group's peers due to Remote instance of NiFi 
is not configured to allow RAW Socket site-to-site communications

这就是为什么我认为这是端口问题

现在使用 Nifi 的默认配置，这可以按预期工作:

SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
            .url("http://localhost:8080/nifi")
            .portName("portNameAsInNifi")
            .transportProtocol(SiteToSiteTransportProtocol.HTTP)
            .requestBatchCount(1)
            .buildConfig();