You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by bbende <gi...@git.apache.org> on 2015/09/30 15:50:18 UTC
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
GitHub user bbende opened a pull request:
https://github.com/apache/flink/pull/1198
FLINK-2740 Adding flink-connector-nifi module with NiFiSource and NiF…
To test with a live NiFi instance, do the following:
* Download the latest 0.3.0 release: https://nifi.apache.org/download.html
* Extract it somewhere and edit conf/nifi.properties to set
nifi.remote.input.socket.port=9081
nifi.remote.input.secure=false
* ./bin/nifi.sh start
* Go to http://localhost:8080/nifi/ in your browser
* Click the templates icon in the top right
* Click Browse and select the template in src/test/resources, click Import
* Drag a new template on to the canvas using the Template icon near the top left
* Press the green Start icon in the toolbar to start all processors
* Run either the NiFiSourceTopology to pull data from NiFi, or the NiFiSinkTopology to push data
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/bbende/flink FLINK-2740
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/1198.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1198
----
commit 3d588a4e6eb36761f3d708e9fd3bce66cfd23a98
Author: Bryan Bende <bb...@hw11977.home>
Date: 2015-09-30T13:15:44Z
FLINK-2740 Adding flink-connector-nifi module with NiFiSource and NiFiSink
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144961452
Ah, you've addressed my feedback. I should refresh my browser before commenting ;)
So +1, good to merge
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40801460
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiDataPacket.java ---
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors.nifi;
+
+import java.util.Map;
+
+/**
+ * <p>
+ * The NiFiDataPacket provides a packaging around a NiFi FlowFile. It wraps both
+ * a FlowFile's content and its attributes so that they can be processed by Flink.
+ * </p>
+ */
+public interface NiFiDataPacket {
+
+ /**
+ * @return the contents of a NiFi FlowFile
+ */
--- End diff --
good catch, checkstyles didn't find that one so I missed it, will update
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144960807
Great. Thank you for your answers. I think its fine to let users deal with cases when multiple systems want to consume data from a NiFi flow.
Once my in-code comments are addressed, the code is good to merge in my opinion.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144423696
The NifiSource-/Topology are just examples, right? If so, could you add Example to the names? We could also unify both into a single example.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144433941
Thank you for your contribution.
You can also check whether the build is passing by executing `mvn clean verify` locally.
Speaking about tests: is there something like a local Nifi runner which starts Nifi within the JVM?
Other systems we integrate with (HDFS, YARN, elasticsearch, kafka) have integration tests that are implemented like that. It would be great to have such a test for the Nifi connector as well.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40797976
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiDataPacket.java ---
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors.nifi;
+
+import java.util.Map;
+
+/**
+ * <p>
+ * The NiFiDataPacket provides a packaging around a NiFi FlowFile. It wraps both
+ * a FlowFile's content and its attributes so that they can be processed by Flink.
+ * </p>
+ */
+public interface NiFiDataPacket {
+
+ /**
+ * @return the contents of a NiFi FlowFile
+ */
+ byte[] getContent();
+
+ /**
+ * @return a Map of attributes that are associated with the NiFi FlowFile
--- End diff --
same as above.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40899440
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/pom.xml ---
@@ -0,0 +1,94 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-connectors-parent</artifactId>
+ <version>0.10-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-connector-nifi</artifactId>
+ <name>flink-connector-nifi</name>
+
+ <packaging>jar</packaging>
+
+ <!-- Allow users to pass custom connector versions -->
+ <properties>
+ <nifi.version>0.3.0</nifi.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-site-to-site-client</artifactId>
+ <version>${nifi.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-tests</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
--- End diff --
Can you clean up the dependencies section a bit to use the same indentation as the other poms?
We are using tabs instead of spaces.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40899251
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiSource.java ---
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors.nifi;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.nifi.remote.Transaction;
+import org.apache.nifi.remote.TransferDirection;
+import org.apache.nifi.remote.client.SiteToSiteClient;
+import org.apache.nifi.remote.client.SiteToSiteClientConfig;
+import org.apache.nifi.remote.protocol.DataPacket;
+import org.apache.nifi.stream.io.StreamUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A source that pulls data from Apache NiFi using the NiFi Site-to-Site client. This source
+ * produces NiFiDataPackets which encapsulate the content and attributes of a NiFi FlowFile.
+ */
+public class NiFiSource extends RichParallelSourceFunction<NiFiDataPacket> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NiFiSource.class);
+
+ private SiteToSiteClient client;
+ private SiteToSiteClientConfig clientConfig;
+ private transient volatile boolean running;
+
+ public NiFiSource(SiteToSiteClientConfig clientConfig) {
+ this.clientConfig = clientConfig;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ client = new SiteToSiteClient.Builder().fromConfig(clientConfig).build();
+ running = true;
+ }
+
+ @Override
+ public void run(SourceContext<NiFiDataPacket> ctx) throws Exception {
+ while (running) {
+ final Transaction transaction = client.createTransaction(TransferDirection.RECEIVE);
--- End diff --
the trxn returned here can be null.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144771981
I realized I didn't fully answer your question about how that works in a cluster... the SiteToSiteClient knows about all the nodes in the NiFi cluster and will pull from the output port on each node. Same concept on the sending side, it will distribute the data to different nodes in the cluster, sending more data to nodes that are considered less busy. The SiteToSiteClient is what we use internally for two NiFi instances/clusters to communicate with each other.
Regarding how they know they belong to the same system, technically they don't, and other non-Flink clients could pull from that same port, but we have to hope that if an organization is using Flink and NiFi they would be ensuring that port was only being used by this specific Flink streaming process, and there could be multiple output ports to support multiple Flink streaming processes pulling.
There is also the option to secure a NiFi instance with SSL, and then provide SSL credentials to the SiteToSite client in order lockdown NiFi so that only authorized clients can pull/push data.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144762615
Thank you for looking into it.
Your analysis was right: I've accidentally put the port into the host parameter `nifi.remote.input.socket.host=9081`.
After I've changed it, everything was working: https://www.youtube.com/watch?v=ylnv5p5Ddk8
In my local setup, it seems that only one of my 8 parallel instances of the source are actually reading data. I suspect that Nifi is delivering data only to one reader.
How would that work in a cluster? How does Nifi know that all the Flink `NifiSource`s connecting to Nifi belong to the same system?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40899230
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiSource.java ---
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors.nifi;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.nifi.remote.Transaction;
+import org.apache.nifi.remote.TransferDirection;
+import org.apache.nifi.remote.client.SiteToSiteClient;
+import org.apache.nifi.remote.client.SiteToSiteClientConfig;
+import org.apache.nifi.remote.protocol.DataPacket;
+import org.apache.nifi.stream.io.StreamUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A source that pulls data from Apache NiFi using the NiFi Site-to-Site client. This source
+ * produces NiFiDataPackets which encapsulate the content and attributes of a NiFi FlowFile.
+ */
+public class NiFiSource extends RichParallelSourceFunction<NiFiDataPacket> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NiFiSource.class);
+
+ private SiteToSiteClient client;
+ private SiteToSiteClientConfig clientConfig;
+ private transient volatile boolean running;
+
+ public NiFiSource(SiteToSiteClientConfig clientConfig) {
+ this.clientConfig = clientConfig;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ client = new SiteToSiteClient.Builder().fromConfig(clientConfig).build();
+ running = true;
+ }
+
+ @Override
+ public void run(SourceContext<NiFiDataPacket> ctx) throws Exception {
+ while (running) {
+ final Transaction transaction = client.createTransaction(TransferDirection.RECEIVE);
+
+ DataPacket dataPacket = transaction.receive();
+ if (dataPacket == null) {
+ transaction.confirm();
+ transaction.complete();
+
+ // no data available. Wait a bit and try again
+ try {
+ Thread.sleep(1000L);
+ } catch (InterruptedException e) {
+
+ }
+ continue;
+ }
+
+ final List<NiFiDataPacket> niFiDataPackets = new ArrayList<>();
+ do {
+ // Read the data into a byte array and wrap it along with the attributes
+ // into a NiFiDataPacket.
+ final InputStream inStream = dataPacket.getData();
+ final byte[] data = new byte[(int) dataPacket.getSize()];
+ StreamUtils.fillBuffer(inStream, data);
+
+ final Map<String, String> attributes = dataPacket.getAttributes();
+
+ niFiDataPackets.add(new StandardNiFiDataPacket(data, attributes));
+ dataPacket = transaction.receive();
+ } while (dataPacket != null);
+
+ // Confirm transaction to verify the data
+ transaction.confirm();
+
+ for (NiFiDataPacket dp : niFiDataPackets) {
+ ctx.collect(dp);
+ }
+
+ transaction.complete();
+ }
--- End diff --
you need to close the `SourceContext` after the while loop
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40899137
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiSink.java ---
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors.nifi;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.nifi.remote.Transaction;
+import org.apache.nifi.remote.TransferDirection;
+import org.apache.nifi.remote.client.SiteToSiteClient;
+import org.apache.nifi.remote.client.SiteToSiteClientConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * A sink that delivers data to Apache NiFi using the NiFi Site-to-Site client. The sink requires
+ * a NiFiDataPacketBuilder which can create instances of NiFiDataPacket from the incoming data.
+ */
+public class NiFiSink<T> extends RichSinkFunction<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NiFiSink.class);
+
+ private SiteToSiteClient client;
+ private SiteToSiteClientConfig clientConfig;
+ private NiFiDataPacketBuilder<T> builder;
+
+ public NiFiSink(SiteToSiteClientConfig clientConfig, NiFiDataPacketBuilder<T> builder) {
+ this.clientConfig = clientConfig;
+ this.builder = builder;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ this.client = new SiteToSiteClient.Builder().fromConfig(clientConfig).build();
+ }
+
+ @Override
+ public void invoke(T value) throws Exception {
+ final NiFiDataPacket niFiDataPacket = builder.createNiFiDataPacket(value, getRuntimeContext());
+
+ final Transaction transaction = client.createTransaction(TransferDirection.SEND);
+ transaction.send(niFiDataPacket.getContent(), niFiDataPacket.getAttributes());
+ transaction.confirm();
+ transaction.complete();
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ try {
+ client.close();
+ } catch (final IOException ioe) {
--- End diff --
I think there is no need to catch this exception here. Flink will also log errors occurring during the close() call.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144446028
Good to know, I'll do a full clean verify on the next update.
Regarding tests, I don't think we have plans to support running an embedded instance of NiFi for full on IT tests. We currently have a mock framework for testing processors with in NiFi, but we could potentially provide a mock implementation of the SiteToSiteClient which would help a lot for these cases, and would be useful for us in general, besides just the Flink integration.
If we were going to provide that, then we would likely need to wait until the next NiFi release (0.4.0) before finalizing this work, unless we incorporated that directly into src/test/java of the flink-connector-nifi module, and removed it later when it was generally available.
Sorry, I mistyped before and meant to say other connectors had topologies in src/main/java, twitter and rabbitmq. So it sounds like it is fine to leave the NiFi topology examples there.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40899178
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiSinkTopologyExample.java ---
@@ -0,0 +1,51 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.flink.streaming.connectors.nifi;
--- End diff --
can you put the two examples into a dedicated examples package?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144432478
I will add Example to the topology class names, do you prefer they live in src/test/resources so they don't get included in the final jar? Wasn't sure because I saw another connector had something similar in src/main/resources.
Also, I suspect the failed build is related to the template file I put in src/test/resources, I will add a license header to that file which hopefully solves that problem.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144461907
Okay, I see. I think it is fine to merge the Nifi connector without a test if there is no infrastructure for automated testing available.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40899307
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiSourceTopologyExample.java ---
@@ -0,0 +1,55 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.flink.streaming.connectors.nifi;
--- End diff --
examples package
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40899297
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiSource.java ---
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors.nifi;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.nifi.remote.Transaction;
+import org.apache.nifi.remote.TransferDirection;
+import org.apache.nifi.remote.client.SiteToSiteClient;
+import org.apache.nifi.remote.client.SiteToSiteClientConfig;
+import org.apache.nifi.remote.protocol.DataPacket;
+import org.apache.nifi.stream.io.StreamUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A source that pulls data from Apache NiFi using the NiFi Site-to-Site client. This source
+ * produces NiFiDataPackets which encapsulate the content and attributes of a NiFi FlowFile.
+ */
+public class NiFiSource extends RichParallelSourceFunction<NiFiDataPacket> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NiFiSource.class);
+
+ private SiteToSiteClient client;
+ private SiteToSiteClientConfig clientConfig;
+ private transient volatile boolean running;
+
+ public NiFiSource(SiteToSiteClientConfig clientConfig) {
+ this.clientConfig = clientConfig;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ client = new SiteToSiteClient.Builder().fromConfig(clientConfig).build();
+ running = true;
+ }
+
+ @Override
+ public void run(SourceContext<NiFiDataPacket> ctx) throws Exception {
+ while (running) {
+ final Transaction transaction = client.createTransaction(TransferDirection.RECEIVE);
+
+ DataPacket dataPacket = transaction.receive();
+ if (dataPacket == null) {
+ transaction.confirm();
+ transaction.complete();
+
+ // no data available. Wait a bit and try again
+ try {
+ Thread.sleep(1000L);
+ } catch (InterruptedException e) {
+
+ }
+ continue;
+ }
+
+ final List<NiFiDataPacket> niFiDataPackets = new ArrayList<>();
+ do {
+ // Read the data into a byte array and wrap it along with the attributes
+ // into a NiFiDataPacket.
+ final InputStream inStream = dataPacket.getData();
+ final byte[] data = new byte[(int) dataPacket.getSize()];
+ StreamUtils.fillBuffer(inStream, data);
+
+ final Map<String, String> attributes = dataPacket.getAttributes();
+
+ niFiDataPackets.add(new StandardNiFiDataPacket(data, attributes));
+ dataPacket = transaction.receive();
+ } while (dataPacket != null);
+
+ // Confirm transaction to verify the data
+ transaction.confirm();
+
+ for (NiFiDataPacket dp : niFiDataPackets) {
+ ctx.collect(dp);
+ }
+
+ transaction.complete();
+ }
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ try {
+ client.close();
+ } catch (final IOException ioe) {
+ LOG.error("Unable to close SiteToSiteClient: " + ioe.getMessage());
--- End diff --
Same here. Flink can handle the Exception
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40797962
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiDataPacket.java ---
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors.nifi;
+
+import java.util.Map;
+
+/**
+ * <p>
+ * The NiFiDataPacket provides a packaging around a NiFi FlowFile. It wraps both
+ * a FlowFile's content and its attributes so that they can be processed by Flink.
+ * </p>
+ */
+public interface NiFiDataPacket {
+
+ /**
+ * @return the contents of a NiFi FlowFile
+ */
--- End diff --
you've used spaces for indentation here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144688632
I followed the tutorial for Nifi that you posted into the pull request. Thanks a lot for the good instructions. I was able to start it and get it to run in a few minutes.
However, when starting the source topology, I got the following error:
```
10/01/2015 12:23:47 Source: Custom Source -> Map -> Sink: Unnamed(2/8) switched to RUNNING
10/01/2015 12:23:47 Source: Custom Source -> Map -> Sink: Unnamed(4/8) switched to FAILED
java.lang.NullPointerException
at org.apache.flink.streaming.connectors.nifi.NiFiSource.run(NiFiSource.java:65)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:57)
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:57)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:198)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:579)
at java.lang.Thread.run(Thread.java:745)
```
it seems that the `final Transaction transaction = client.createTransaction(TransferDirection.RECEIVE);` call is returning null.
This is what my NiFi screen looks like: http://i.imgur.com/FDFgrdu.png
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1198#discussion_r40899210
--- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-nifi/src/main/java/org/apache/flink/streaming/connectors/nifi/NiFiSource.java ---
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors.nifi;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.nifi.remote.Transaction;
+import org.apache.nifi.remote.TransferDirection;
+import org.apache.nifi.remote.client.SiteToSiteClient;
+import org.apache.nifi.remote.client.SiteToSiteClientConfig;
+import org.apache.nifi.remote.protocol.DataPacket;
+import org.apache.nifi.stream.io.StreamUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A source that pulls data from Apache NiFi using the NiFi Site-to-Site client. This source
+ * produces NiFiDataPackets which encapsulate the content and attributes of a NiFi FlowFile.
+ */
+public class NiFiSource extends RichParallelSourceFunction<NiFiDataPacket> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NiFiSource.class);
+
+ private SiteToSiteClient client;
+ private SiteToSiteClientConfig clientConfig;
+ private transient volatile boolean running;
+
+ public NiFiSource(SiteToSiteClientConfig clientConfig) {
+ this.clientConfig = clientConfig;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ client = new SiteToSiteClient.Builder().fromConfig(clientConfig).build();
+ running = true;
+ }
+
+ @Override
+ public void run(SourceContext<NiFiDataPacket> ctx) throws Exception {
+ while (running) {
+ final Transaction transaction = client.createTransaction(TransferDirection.RECEIVE);
+
+ DataPacket dataPacket = transaction.receive();
+ if (dataPacket == null) {
+ transaction.confirm();
+ transaction.complete();
+
+ // no data available. Wait a bit and try again
+ try {
+ Thread.sleep(1000L);
--- End diff --
I think we should make this configurable for users with low latency requirements, ...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144434371
Which connector has the examples in `src/main/resources` ? I think its okay to put the examples into the src/main/java folder.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-145400455
I'm going to merge this PR ...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144759397
Working on updates to address the feedback... Regarding the NullPointerException, I'm going to protect against that in the code, but I'm still trying to think of why you ran into that.
Did you set nifi.remote.input.socket.port in nifi.properties?
I was able to produce that NPE when I did not set this property.
Could you also see if any errors are showing up in NIFI_HOME/logs/nifi-app.log at the same time?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/1198
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144769010
I think the example data flow in NiFi is producing data fairly slowly, this is based on the scheduling tab of the GenerateFlowFile processor being set to 2 secs. We could set this to 0 sec to start really going by stopping the processor, right-clicking and selecting Configure.
The SiteToSiteClientConfig has options to set batch count but the example isn't setting it so it is using some default, and the first instance of the source probably runs and pulls all the available data, and when the second instance runs nothing is left.
If you set the batch count lower in the example, like this:
SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
.url("http://localhost:8080/nifi")
.portName("Data for Flink")
.requestBatchCount(5)
.buildConfig();
and then have more than 5 flow files queued in NiFi when you start the source topology, I think the other instances will start pulling as well. This seemed to work for me using parallelism of 2, but of course let me know if you are seeing something different.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: FLINK-2740 Adding flink-connector-nifi module ...
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the pull request:
https://github.com/apache/flink/pull/1198#issuecomment-144776291
Each Flink streaming job/process would pull from a different port, the port name is specified in the SiteToSiteClientConfig for the source:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---