魔术头球没有出现

Question

我正在尝试使用这个例子将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)作为URL添加到我的SiteToSiteClientConfig和http://localhost:8080/nifi中。

首先，我更改端口的原因是，如果不指定协议HTTP，默认情况下它将使用RAW。当从Flink端使用RAW协议时，客户端无法创建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();