nifi-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From markap14 <...@git.apache.org>
Subject [GitHub] nifi pull request: NIFI-1192 added support for dynamic properties ...
Date Tue, 24 Nov 2015 01:38:58 GMT
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.
---

Mime
View raw message