Hi Bobby, Your understanding is correct that @OnScheduled will be called when the user clicks Start (or upon restart of nifi if it was running when shutdown). The onTrigger method will be called continually. It's hard to say why it performs poorly with just a quick description of the Processor.
It's possible that the Processor is 'yielding', which would cause it to not be triggered for 1 second (by default). If that's the case, then you could either avoid calling ProcessContext.yield() or you can configure the Processor's Settings tab to set Yield Duration to "0 secs" instead of "1 sec". This would also be evident by the amount of time that the Processor is running. If the Processor always shows the maximum number of threads running, then it means that Processor itself may be struggling, but if it typically does not show all of the threads running, this likely indicates that the Processor is Yielding, or that you need to allocate more threads to the Controller (by default, all Processors share a thread pool that has only 10 threads. You can control that by going to the Hamburger menu in the top-right corner and going to Controller Settings and changing the Max number of Timer-Driven Threads to a larger value). If the problem is in the Processor itself, then I would recommend using JVisualVM (packed with the JDK) to "Sample" the application or use YourKit to profile it. Hope this is helpful. -Mark > On Jul 30, 2018, at 4:38 AM, Bobby <[email protected]> wrote: > > Hello, i am trying to create custom processor of listen tcp; My requirements > are: > > The source system will create a connection and set a session; After a > session established, a message will be sent then my processor need to decode > the message, based on that decoded output, i need to create a response back > using the same socket; After reply has been sent, the socket will be closed > by the source system; > > See, the client access my processor in one socket session one message style; > I can't use the built in ListenTCP since it doesn't have custom reply > response in same session (more specifically, same output stream) therefore i > need to make one. > > My class structure is: > > *ListenTCP - extends abstractProcessor* > This is the main processor, will instantiate and call ConcurrentListener() > > *ConcurrentListener* > Setup and start the server socket and accept connection > > *ConcurrentAdapterHandler* > Wrapper class for executing runnable ConcurentAdapter only, the idea is to > process the message in another thread and store Future object in a > concurrentLinkedQueue here; The Queue will be consumed in onTrigger() in > main processor class. > > *ConcurrentAdapter* > Decode and send response back > > In my ListenTCP i implemented: > > *@OnScheduled* > OnSchedule() > - Create and initialize server socket in separate runnable class and using > separate daemon thread to run it > > OnTrigger() > - Get queue in class ConcurrentListener > - If queue > 0 then session.create() for createing new flow file and > session.transfer it else return; > > The problem is, the message processed rate is very low compared with the > same code implemented in Spring Java style. Where is my mistake? My > understanding is: > onScheduled will be called once for every time i hit the start button at > nifi GUI > onTrigger will be called repeteadly and simultanously (based on how many > thread assigned in Nifi GUI) > > > > > > > ----- > > ----------------- > Bobby > -- > Sent from: http://apache-nifi-developer-list.39713.n7.nabble.com/
