[
https://issues.apache.org/jira/browse/NIFI-1827?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15264427#comment-15264427
]
Christopher McDermott commented on NIFI-1827:
---------------------------------------------
I'm not exactly sure how things got in this situation but stepping through the
following code in the debugger ({{RoundRobinPartitioner}}) I see that
{{index}} has become greater than {{numberOfPartitons}} and
{{numberOfPartitions}} has a correct value.
{code}
private int next(int numberOfPartitions) {
if (index == numberOfPartitions) {
index = 0;
}
int indexToReturn = index++;
return indexToReturn;
}
{code}
This causes {{index}} to never reset to {{0}}.
> PutKafka attempts to write to non-existent partition.
> ------------------------------------------------------
>
> Key: NIFI-1827
> URL: https://issues.apache.org/jira/browse/NIFI-1827
> Project: Apache NiFi
> Issue Type: Bug
> Affects Versions: 0.7.0
> Reporter: Christopher McDermott
>
> PutKafka attempts to write to non-existent partition. I have not verified
> yet but I think the problem can be triggered by deleting a topic while the
> processors is running, and then recreating the topic with the same name.
> Since the problem has occurred I have not been able to make it go away. I've
> recreated the processor in the flow but the new processor exhibits the same
> behavior.
> 2016-04-29 12:00:53,550 ERROR [Timer-Driven Process Thread-1]
> o.apache.nifi.processors.kafka.PutKafka
> java.lang.IllegalArgumentException: Invalid partition given with record: 4 is
> not in the range [0...1].
> at
> org.apache.kafka.clients.producer.internals.Partitioner.partition(Partitioner.java:52)
> ~[na:na]
> at
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:333)
> ~[na:na]
> at
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:248)
> ~[na:na]
> at
> org.apache.nifi.processors.kafka.KafkaPublisher.toKafka(KafkaPublisher.java:203)
> ~[na:na]
> at
> org.apache.nifi.processors.kafka.KafkaPublisher.publish(KafkaPublisher.java:137)
> ~[na:na]
> at
> org.apache.nifi.processors.kafka.PutKafka$1.process(PutKafka.java:300)
> ~[na:na]
> at
> org.apache.nifi.controller.repository.StandardProcessSession.read(StandardProcessSession.java:1807)
> ~[nifi-framework-core-0.7.0-SNAPSHOT.jar:0.7.0-SNAPSHOT]
> at
> org.apache.nifi.controller.repository.StandardProcessSession.read(StandardProcessSession.java:1778)
> ~[nifi-framework-core-0.7.0-SNAPSHOT.jar:0.7.0-SNAPSHOT]
> at
> org.apache.nifi.processors.kafka.PutKafka.onTrigger(PutKafka.java:296)
> ~[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]
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> [na:1.8.0_45]
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> [na:1.8.0_45]
> at java.lang.Thread.run(Thread.java:745) [na:1.8.0_45]
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
