You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@nifi.apache.org by olegz <gi...@git.apache.org> on 2015/11/23 21:16:35 UTC
[GitHub] nifi pull request: NIFI-1192 added support for dynamic properties ...
GitHub user olegz opened a pull request:
https://github.com/apache/nifi/pull/131
NIFI-1192 added support for dynamic properties to GetKafka
Due to the fact that current component uses artificial names for properties set via UI and then maps those properties to the actual names used by Kafka, we can not rely on NiFi UI to display an error if user attempts to set a dynamic property which will eventually map to the same Kafka property. So, I’ve decided that any dynamic property will simply override an existing property with WARNING message displayed. It is actually consistent with how Kafka does it and displayed the overrides in the console. Updated the relevant annotation description.
It is also worth to mentioned that current code was using an old property from Kafka 0.7 (“zk.connectiontimeout.ms”) which is no longer present in Kafka 0.8 (WARN Timer-Driven Process Thread-7 utils.VerifiableProperties:83 - Property zk.connectiontimeout.ms is not valid). The add/override strategy would provide for more flexibility when dealing with Kafka volatile configuration until things will settle down and we can get some sensible defaults in place.
While doing it addressed the following issues that were discovered while making modification and testing:
ISSUE: When GetKafka started and there are no messages in Kafka topic the onTrigger(..) method would block due to the fact that Kafka’s ConsumerIterator.hasNext() blocks. When attempt was made to stop GetKafka would stops successfully due to the interrupt. However in UI it would appear as ERROR based on the fact that InterruptException was not handled.
RESOLUTION: After discussing it with @markap14 the the general desire is to let the task exit as quick as possible and that the whole thread maintenance logic was there initially due to the fact that there was no way to tell Kafka consumer to return immediately if there are no events. In this patch we are now using ‘consumer.timeout.ms’ property of Kafka and setting its value to 1 millisecond (default is -1 - always block infinitely). This ensures that tasks that attempted to read an empty topic will exit immediately just to be rescheduled by NiFi based on user configurations.
ISSUE: Kafka would not release FlowFile with events if it didn’t have enough to complete the batch since it would block waiting for more messages (based on the blocking issue described above).
RESOLUTION: The invocation of hasNext() results in Kafka’s ConsumerTimeoutException which is handled in the catch block where the FlowFile with partial batch will be released to success. Not sure if we need to put a WARN message. In fact in my opinion we should not as it may create unnecessary confusion.
ISSUE: When configuring a consumer for topic and specifying multiple concurrent consumers in ‘topicCountMap’ based on 'context.getMaxConcurrentTasks()’ each consumer would bind to a topic partition. If you have less partitions then the value returned by 'context.getMaxConcurrentTasks()’ you would essentially allocate Kafka resources that would never get a chance to receive a single message (see more here https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example).
RESOLUTION: Logic was added to determine the amount of partitions for a topic and in the event where 'context.getMaxConcurrentTasks()’ value is greater than the amount of partitions, the partition count will be used to when creating ‘topicCountMap’ and WARNING message will be displayed)see code). Unfortunately we can’t do anything with the actual tasks, but based on current state of the code they will exit immediately just to be rescheduled where the process will repeat. NOTE: That is not ideal as it will be rescheduling tasks that will never have a chance to do anything, but at least it could be fixed on the user side after reading the warning message.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/olegz/nifi NIFI-1192B
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/131.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 #131
----
commit f9d1b2811a08c372baf660a8a5e8d0f73e1a23a2
Author: Oleg Zhurakousky <ol...@suitcase.io>
Date: 2015-11-23T20:15:03Z
NIFI-1192 added support for dynamic properties to GetKafka
Due to the fact that current component uses artificial names for properties set via UI and then maps those properties to the actual names used by Kafka, we can not rely on NiFi UI to display an error if user attempts to set a dynamic property which will eventually map to the same Kafka property. So, I’ve decided that any dynamic property will simply override an existing property with WARNING message displayed. It is actually consistent with how Kafka does it and displayed the overrides in the console. Updated the relevant annotation description.
It is also worth to mentioned that current code was using an old property from Kafka 0.7 (“zk.connectiontimeout.ms”) which is no longer present in Kafka 0.8 (WARN Timer-Driven Process Thread-7 utils.VerifiableProperties:83 - Property zk.connectiontimeout.ms is not valid). The add/override strategy would provide for more flexibility when dealing with Kafka volatile configuration until things will settle down and we can get some sensible defaults in place.
While doing it addressed the following issues that were discovered while making modification and testing:
ISSUE: When GetKafka started and there are no messages in Kafka topic the onTrigger(..) method would block due to the fact that Kafka’s ConsumerIterator.hasNext() blocks. When attempt was made to stop GetKafka would stops successfully due to the interrupt. However in UI it would appear as ERROR based on the fact that InterruptException was not handled.
RESOLUTION: After discussing it with @markap14 the the general desire is to let the task exit as quick as possible and that the whole thread maintenance logic was there initially due to the fact that there was no way to tell Kafka consumer to return immediately if there are no events. In this patch we are now using ‘consumer.timeout.ms’ property of Kafka and setting its value to 1 millisecond (default is -1 - always block infinitely). This ensures that tasks that attempted to read an empty topic will exit immediately just to be rescheduled by NiFi based on user configurations.
ISSUE: Kafka would not release FlowFile with events if it didn’t have enough to complete the batch since it would block waiting for more messages (based on the blocking issue described above).
RESOLUTION: The invocation of hasNext() results in Kafka’s ConsumerTimeoutException which is handled in the catch block where the FlowFile with partial batch will be released to success. Not sure if we need to put a WARN message. In fact in my opinion we should not as it may create unnecessary confusion.
ISSUE: When configuring a consumer for topic and specifying multiple concurrent consumers in ‘topicCountMap’ based on 'context.getMaxConcurrentTasks()’ each consumer would bind to a topic partition. If you have less partitions then the value returned by 'context.getMaxConcurrentTasks()’ you would essentially allocate Kafka resources that would never get a chance to receive a single message (see more here https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example).
RESOLUTION: Logic was added to determine the amount of partitions for a topic and in the event where 'context.getMaxConcurrentTasks()’ value is greater than the amount of partitions, the partition count will be used to when creating ‘topicCountMap’ and WARNING message will be displayed)see code). Unfortunately we can’t do anything with the actual tasks, but based on current state of the code they will exit immediately just to be rescheduled where the process will repeat. NOTE: That is not ideal as it will be rescheduling tasks that will never have a chance to do anything, but at least it could be fixed on the user side after reading the warning message.
----
---
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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/131#discussion_r45688132
--- Diff: nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/processors/kafka/TestGetKafka.java ---
@@ -45,13 +45,18 @@ public static void configureLogging() {
}
@Test
- @Ignore("Intended only for local tests to verify functionality.")
public void testIntegrationLocally() {
+ EmbeddedKafka kafka = new EmbeddedKafka();
+ kafka.start();
+
+ kafka.addTopic("testX", 1);
+ kafka.sendMessage("testX");
+
final TestRunner runner = TestRunners.newTestRunner(GetKafka.class);
- runner.setProperty(GetKafka.ZOOKEEPER_CONNECTION_STRING, "192.168.0.101:2181");
+ runner.setProperty(GetKafka.ZOOKEEPER_CONNECTION_STRING, "localhost:2181");
--- End diff --
If we are running an embedded service here, can we use a port of 0? Expecting to open a specific port is dangerous in unit tests and often results in unit test failures.
---
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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on a diff in the pull request:
https://github.com/apache/nifi/pull/131#discussion_r45691543
--- Diff: nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/processors/kafka/TestGetKafka.java ---
@@ -45,13 +45,18 @@ public static void configureLogging() {
}
@Test
- @Ignore("Intended only for local tests to verify functionality.")
public void testIntegrationLocally() {
+ EmbeddedKafka kafka = new EmbeddedKafka();
+ kafka.start();
+
+ kafka.addTopic("testX", 1);
+ kafka.sendMessage("testX");
+
final TestRunner runner = TestRunners.newTestRunner(GetKafka.class);
- runner.setProperty(GetKafka.ZOOKEEPER_CONNECTION_STRING, "192.168.0.101:2181");
+ runner.setProperty(GetKafka.ZOOKEEPER_CONNECTION_STRING, "localhost:2181");
--- End diff --
I don't think this is intended to be automatically run as part of the final version. @olegz just make sure to remember to re-add the @Ignore before the final 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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the pull request:
https://github.com/apache/nifi/pull/131#issuecomment-159052094
@trkurc @markap14 @joewitt
Guys, this one is strictly for review as I am still working on adding the same dynamic properties logic for PutKafka. But since it contains somewhat significant changes, it would be worth taking a look. Commit message has all the details.
Cheers
---
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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/131#discussion_r45687929
--- Diff: nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/GetKafka.java ---
@@ -203,26 +205,69 @@
return relationships;
}
- @OnScheduled
public void createConsumers(final ProcessContext context) {
final String topic = context.getProperty(TOPIC).getValue();
- final Map<String, Integer> topicCountMap = new HashMap<>(1);
- topicCountMap.put(topic, context.getMaxConcurrentTasks());
-
final Properties props = new Properties();
props.setProperty("zookeeper.connect", context.getProperty(ZOOKEEPER_CONNECTION_STRING).getValue());
props.setProperty("group.id", context.getProperty(GROUP_ID).getValue());
props.setProperty("client.id", context.getProperty(CLIENT_NAME).getValue());
props.setProperty("auto.commit.interval.ms", String.valueOf(context.getProperty(ZOOKEEPER_COMMIT_DELAY).asTimePeriod(TimeUnit.MILLISECONDS)));
- props.setProperty("auto.commit.enable", "true"); // just be explicit
props.setProperty("auto.offset.reset", context.getProperty(AUTO_OFFSET_RESET).getValue());
- props.setProperty("zk.connectiontimeout.ms", context.getProperty(ZOOKEEPER_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).toString());
+ props.setProperty("zookeeper.connection.timeout.ms", context.getProperty(ZOOKEEPER_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).toString());
props.setProperty("socket.timeout.ms", context.getProperty(KAFKA_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).toString());
+ for (final Entry<PropertyDescriptor, String> entry : context.getProperties().entrySet()) {
+ PropertyDescriptor descriptor = entry.getKey();
+ if (descriptor.isDynamic()) {
+ if (props.containsKey(descriptor.getName())) {
+ this.getLogger().warn("Overriding existing property '" + descriptor.getName() + "' which had value of '"
+ + props.getProperty(descriptor.getName()) + "' with dynamically set value '"
+ + entry.getValue() + "'.");
+ }
+ props.setProperty(descriptor.getName(), entry.getValue());
+ }
+ }
+
+ /*
+ * Unless user sets it to some explicit value we are setting it to the
+ * lowest possible value of 1 millisecond to ensure the
+ * consumerStream.hasNext() doesn't block. See
+ * http://kafka.apache.org/documentation.html#configuration) as well as
+ * comment in 'catch ConsumerTimeoutException' in onTrigger() for more
+ * explanation as to the reasoning behind it.
+ */
+ if (!props.containsKey("consumer.timeout.ms")) {
+ this.getLogger().info("Setting 'consumer.timeout.ms' to 1 milliseconds to avoid consumer"
+ + " block in the event when no events are present in Kafka topic. If you wish to change this value "
+ + " set it as dynamic property. If you wish to explicitly enable consumer block (at yoru own risk)"
--- End diff --
type here: says "yoru" instead of "your"
---
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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/131#discussion_r45688059
--- Diff: nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/processors/kafka/TestGetKafka.java ---
@@ -45,13 +45,18 @@ public static void configureLogging() {
}
@Test
- @Ignore("Intended only for local tests to verify functionality.")
public void testIntegrationLocally() {
+ EmbeddedKafka kafka = new EmbeddedKafka();
+ kafka.start();
+
+ kafka.addTopic("testX", 1);
+ kafka.sendMessage("testX");
+
final TestRunner runner = TestRunners.newTestRunner(GetKafka.class);
- runner.setProperty(GetKafka.ZOOKEEPER_CONNECTION_STRING, "192.168.0.101:2181");
+ runner.setProperty(GetKafka.ZOOKEEPER_CONNECTION_STRING, "localhost:2181");
runner.setProperty(GetKafka.TOPIC, "testX");
- runner.setProperty(GetKafka.KAFKA_TIMEOUT, "3 secs");
- runner.setProperty(GetKafka.ZOOKEEPER_TIMEOUT, "3 secs");
+ // runner.setProperty(GetKafka.KAFKA_TIMEOUT, "3 secs");
--- End diff --
If this line and the next are not needed, can you remove them rather than commenting them out?
---
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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the pull request:
https://github.com/apache/nifi/pull/131#issuecomment-159345927
@markap14 @trkurc The PR comments were addressed. I was hoping to get Kafka embedded server with this commit as well (for testing), but so far it doesn't appear to be very stable. We can do it by the next release.
---
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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/131
---
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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on the pull request:
https://github.com/apache/nifi/pull/131#issuecomment-159124855
@olegz I left a handful of comments in-line. Otherwise, looks great! Very happy to have someone else digging into these Processors and the code is certainly looking cleaner now that you've refactored it a bit to use the newer client. Thanks!!
---
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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/131#discussion_r45687909
--- Diff: nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/GetKafka.java ---
@@ -70,6 +70,10 @@
+ " the message does not have a key, or if the batch size is greater than 1, this attribute will not be added"),
@WritesAttribute(attribute = "kafka.partition", description = "The partition of the Kafka Topic from which the message was received. This attribute is added only if the batch size is 1"),
@WritesAttribute(attribute = "kafka.offset", description = "The offset of the message within the Kafka partition. This attribute is added only if the batch size is 1")})
+@DynamicProperty(name = "The name of a Kafka configuration property.", value = "The value of a given Kafka configuration property.", description = "These properties will be added on the Kafka configuration after loading any provided configuration properties."
--- End diff --
@olegz this line is 260 characters; fails contrib-check. Can you break that line apart?
---
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] nifi pull request: NIFI-1192 added support for dynamic properties ...
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the pull request:
https://github.com/apache/nifi/pull/131#issuecomment-159129775
Thanks @markap14! indeed it's a bit dirty and I have to polish it for style-check, comments etc. As I mentioned in commit message somewhere this was primarily for initial review to ensure that someone else can take a quick look and make sure I didn't go off the rails 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.
---