[
https://issues.apache.org/jira/browse/NIFI-4724?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16313331#comment-16313331
]
ASF subversion and git services commented on NIFI-4724:
-------------------------------------------------------
Commit e5ed62a98fe0c15fcba865edfe00eac673a6ac51 in nifi's branch
refs/heads/master from [~ijokarumawak]
[ https://git-wip-us.apache.org/repos/asf?p=nifi.git;h=e5ed62a ]
NIFI-4724: Support 0 byte message with PublishKafka
Before this fix, PublishKafka (0.9) and PublishKafka_0_10 fail with empty
incoming FlowFiles due to 'transfer relationship not specified' error.
Because the internal 'publish' method is not called as StreamDemarcator does
not emit any token regardless whether demarcator is set or not.
As for PublishKafka_0_11 and PublishKafka_1_0, empty FlowFiles are transferred
to 'success' relationship, however no Kafka message is sent to Kafka.
Since Kafka allows 0 byte body empty messages, NiFi should be able to send it,
too.
This commit changes above current situation to the followings, with all
PublishKafka_* processors:
- If demarcator is not set, then publish incoming FlowFile content as it is.
This enables sending an empty Kafka message.
- If demarcator is set, send each token as a separate message.
Even if no token is found (empty incoming FlowFile), transfer the FlowFile to
'success'.
This closes #2362.
Signed-off-by: Mark Payne <[email protected]>
> Publish kafka processors fails with FlowFileHandlingException if the flow
> file is empty
> ---------------------------------------------------------------------------------------
>
> Key: NIFI-4724
> URL: https://issues.apache.org/jira/browse/NIFI-4724
> Project: Apache NiFi
> Issue Type: Bug
> Components: Extensions
> Affects Versions: 1.1.0
> Reporter: Mahesh Nayak
> Assignee: Koji Kawamura
> Fix For: 1.5.0
>
>
> 1. Construct the flow GenerateFlowFile --> PublishKafka --> PutFile
> 2. In GenerateFlowFile set the "File Size" to 0B.
> 3. Start the flow.
> Result : Kafka processor throws below exception
> {code:None}
> 2017-12-27 02:49:21,933 WARN [Timer-Driven Process Thread-9]
> o.a.n.c.t.ContinuallyRunProcessorTask Administratively Yielding
> PublishKafka_0_10[id=95dbc77a-0160-1000-0000-000069761c4e] due to uncaught
> Exception: org.apache.nifi.processor.exception.FlowFileHandlingException:
> StandardFlowFileRecord[uuid=4d6cb989-b6a7-4129-9dfc-1598ee2b3937,claim=,offset=0,name=7061091998478433,size=0]
> transfer relationship not specified
> 2017-12-27 02:49:21,933 WARN [Timer-Driven Process Thread-9]
> o.a.n.c.t.ContinuallyRunProcessorTask
> org.apache.nifi.processor.exception.FlowFileHandlingException:
> StandardFlowFileRecord[uuid=4d6cb989-b6a7-4129-9dfc-1598ee2b3937,claim=,offset=0,name=7061091998478433,size=0]
> transfer relationship not specified
> at
> org.apache.nifi.controller.repository.StandardProcessSession.checkpoint(StandardProcessSession.java:251)
> at
> org.apache.nifi.controller.repository.StandardProcessSession.commit(StandardProcessSession.java:321)
> at
> org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:28)
> at
> org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1120)
> at
> org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:147)
> at
> org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:47)
> at
> org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:128)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
> at
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
> at
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
