nifi-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "McDermott, Chris Kevin (MSDU - STaTS/StorefrontRemote)" <chris.mcderm...@hpe.com>
Subject Re: NPE in PutKafka processors
Date Wed, 13 Apr 2016 20:23:59 GMT
Here is a probable patch. I’ve tested it and it seems to do the trick.

diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/PutKafka.java
b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/PutKafka.java
index 3b5eb4f..19e06ea 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/PutKafka.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/PutKafka.java
@@ -1,4 +1,5 @@
 /*
+
  * 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.
@@ -393,7 +394,7 @@ public class PutKafka extends AbstractProcessor {
         attributes.put(ATTR_FAILED_SEGMENTS, new String(failedSegments.toByteArray(), StandardCharsets.UTF_8));
         attributes.put(ATTR_TOPIC, messageContext.getTopicName());
         attributes.put(ATTR_KEY, messageContext.getKeyBytesAsString());
-        attributes.put(ATTR_DELIMITER, new String(messageContext.getDelimiterBytes(), StandardCharsets.UTF_8));
+        attributes.put(ATTR_DELIMITER, messageContext.getDelimeterBytesAsString());
         return attributes;
     }

diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/SplittableMessageContext.java
b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/SplittableMessageContext.java
index d5f1c0b..4d63b7b 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/SplittableMessageContext.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/SplittableMessageContext.java
@@ -16,6 +16,7 @@
  */
 package org.apache.nifi.processors.kafka;

+import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.util.BitSet;

@@ -108,6 +109,13 @@ final class SplittableMessageContext {
     }

     /**
+     * Returns the delimiter bytes as String
+     */
+    String getDelimeterBytesAsString() {
+        return this.delimiterBytes != null ? new String(this.delimiterBytes, StandardCharsets.UTF_8)
: null;
+    }
+
+    /**
      * Returns the key bytes as String
      */
     String getKeyBytesAsString() {







On 4/13/16, 2:54 PM, "McDermott, Chris Kevin (MSDU - STaTS/StorefrontRemote)" <chris.mcdermott@hpe.com>
wrote:

>Hi!
>
>I’m running a 0.7.0 snapshot version and am running into an NPE with the PutKafka processor.
>
>I’m thinking Oleg would be interested in this.
>
>java.lang.NullPointerException: null
>at java.lang.String.<init>(String.java:503) ~[na:1.8.0_45]
>at org.apache.nifi.processors.kafka.PutKafka.buildFailedFlowFileAttributes(PutKafka.java:396)
~[na:na]
>at org.apache.nifi.processors.kafka.PutKafka.onTrigger(PutKafka.java:308) ~[na:na]
>at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27) ~[nifi-api-0.7.0-SNAPSHOT.jar:0.7.0-SNAPSHOT]
>at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1059)
~[nifi-framework-core-0.7.0-SNAPSHOT.jar:0.7.0-SNAPSHOT]
>at org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:136)
[nifi-framework-core-0.7.0-SNAPSHOT.jar:0.7.0-SNAPSHOT]
>at org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:47)
[nifi-framework-core-0.7.0-SNAPSHOT.jar:0.7.0-SNAPSHOT]
>at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:123)
[nifi-framework-core-0.7.0-SNAPSHOT.jar:0.7.0-SNAPSHOT]
>at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_45]
>at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) [na:1.8.0_45]
>at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
[na:1.8.0_45]
>at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
[na:1.8.0_45]
>
>Thanks,
>
>Chris
Mime
View raw message