My JIRA id is saimishra Regards Sai
On Tue, Oct 25, 2016 at 10:55 AM, Matthias J. Sax <matth...@confluent.io> wrote: > -----BEGIN PGP SIGNED MESSAGE----- > Hash: SHA512 > > What is your JIRA ID? We can add you to the contributor list to give > you permission. > > - -Matthias > > > On 10/25/16 10:48 AM, saiprasad mishra wrote: > > Hi Matthias Thanks for the reply. I think I don't have permission > > for this. If you can grant me permission I can create one (my > > handle is saimishra). Or you can go ahead and create one > > > > I may need permission to create JIRA as I might report more issues > > after discussing with you over here. > > > > Regards Sai > > > > On Tue, Oct 25, 2016 at 10:26 AM, Matthias J. Sax > > <matth...@confluent.io> wrote: > > > > Hi, > > > > sorry for late reply. Seems like a bug to me; within > > Processor#process() accessing the context should work. Can you open > > a JIRA for it? > > > > -Matthias > > > > On 10/23/16 10:28 PM, saiprasad mishra wrote: > >>>> Sorry for the email again > >>>> > >>>> I was expecting it to work always when accessed from > >>>> process() method as this corresponds to each kafka > >>>> message/record processing. I understand illegalstate by the > >>>> time punctuate() is called as its already batched by time > >>>> interval > >>>> > >>>> Regards Sai > >>>> > >>>> On Sun, Oct 23, 2016 at 9:18 PM, saiprasad mishra > >>>> <saiprasadmis...@gmail.com > >>>>> wrote: > >>>> > >>>>> Hi > >>>>> > >>>>> his is with my streaming app kafka 10.1.0. > >>>>> > >>>>> My flow looks something like below > >>>>> > >>>>> source topic stream -> filter for null value ->map to make > >>>>> it keyed by id ->custom processor to mystore -> to another > >>>>> topic -> ktable > >>>>> > >>>>> I am hitting the below type of exception in a custom > >>>>> processor class if I try to access offset() or partition() > >>>>> or timestamp() from the ProcessorContext in the process() > >>>>> method. I was hoping it would return the partition and > >>>>> offset for the enclosing topic(in this case source topic) > >>>>> where its consuming from or -1 based on the api docs. > >>>>> > >>>>> Looks like only in certain cases it is accessible. is it > >>>>> getting lost in transformation phases. > >>>>> > >>>>> Same issue happens on if i try to access them in > >>>>> punctuate() method but some where I saw that it might not > >>>>> work in punctuate(). Any reason for this or any link > >>>>> describing this will be helpful > >>>>> > >>>>> > >>>>> ================================================================== > == > >>>>> > >>>>> > >>>>> > >>>>> > > > >>>>> > java.lang.IllegalStateException: This should not happen as offset() > >>>>> should only be called while a record is processed at > >>>>> org.apache.kafka.streams.processor.internals. > >>>>> ProcessorContextImpl.offset(ProcessorContextImpl.java:181) > >>>>> ~[kafka-streams-0.10.1.0.jar!/:?] at > >>>>> com.sai.repo.MyStore.process(MyStore.java:72) > >>>>> ~[classes!/:?] at > >>>>> com.sai.repo.MyStore.process(MyStore.java:39) > >>>>> ~[classes!/:?] at > >>>>> org.apache.kafka.streams.processor.internals.ProcessorNode.process > (Pr > > > >>>>> > ocessorNode.java:82) > >>>>> > >>>>> > >>>>> > > ~[kafka-streams-0.10.1.0.jar!/:?] > >>>>> at org.apache.kafka.streams.processor.internals. > >>>>> ProcessorContextImpl.forward(ProcessorContextImpl.java:204) > >>>>> > >>>>> > ~[kafka-streams-0.10.1.0.jar!/:?] at > >>>>> org.apache.kafka.streams.kstream.internals.KStreamMap$ > >>>>> KStreamMapProcessor.process(KStreamMap.java:43) > >>>>> ~[kafka-streams-0.10.1.0.jar!/:?] at > >>>>> org.apache.kafka.streams.processor.internals.ProcessorNode.process > (Pr > > > >>>>> > ocessorNode.java:82) > >>>>> > >>>>> > >>>>> > > ~[kafka-streams-0.10.1.0.jar!/:?] > >>>>> at org.apache.kafka.streams.processor.internals. > >>>>> ProcessorContextImpl.forward(ProcessorContextImpl.java:204) > >>>>> > >>>>> > ~[kafka-streams-0.10.1.0.jar!/:?] at > >>>>> org.apache.kafka.streams.kstream.internals.KStreamFilter$ > >>>>> KStreamFilterProcessor.process(KStreamFilter.java:44) > >>>>> ~[kafka-streams-0.10.1.0.jar!/:?] at > >>>>> org.apache.kafka.streams.processor.internals.ProcessorNode.process > (Pr > > > >>>>> > ocessorNode.java:82) > >>>>> > >>>>> > >>>>> > > ~[kafka-streams-0.10.1.0.jar!/:?] > >>>>> at org.apache.kafka.streams.processor.internals. > >>>>> ProcessorContextImpl.forward(ProcessorContextImpl.java:204) > >>>>> > >>>>> > ~[kafka-streams-0.10.1.0.jar!/:?] at > >>>>> org.apache.kafka.streams.processor.internals. > >>>>> SourceNode.process(SourceNode.java:66) > >>>>> ~[kafka-streams-0.10.1.0.jar!/:?] at > >>>>> org.apache.kafka.streams.processor.internals. > >>>>> StreamTask.process(StreamTask.java:181) > >>>>> ~[kafka-streams-0.10.1.0.jar!/:?] at > >>>>> org.apache.kafka.streams.processor.internals.StreamThread.runLoop( > Str > > > >>>>> > eamThread.java:436) > >>>>> > >>>>> > >>>>> > > ~[kafka-streams-0.10.1.0.jar!/:?] > >>>>> at org.apache.kafka.streams.processor.internals. > >>>>> StreamThread.run(StreamThread.java:242) > >>>>> [kafka-streams-0.10.1.0.jar!/:?] > >>>>> ================================================================== > === > >>>>> > >>>>> > >>>>> > >>>>> > >>>>> > > > >>>>> > Regards > >>>>> Sai > >>>>> > >>>> > >> > > > -----BEGIN PGP SIGNATURE----- > Comment: GPGTools - https://gpgtools.org > > iQIcBAEBCgAGBQJYD5yXAAoJECnhiMLycopP+Y4QAK8vkvLauOiJQrb2b56bwZIG > 7uSMnCagjGYnpMnpI1pB88ccJhEMgCvKI8NctXzv967oRoNsRAREdbmF/Li3Vfot > u/Qkk4OUHQzHVu+vn+jEoIG3wZ2d6I40lQWKewJj7YAFhFgLcXLDO9qb447UMJgf > SNI30pmVXb0s6jrF2y20x/GLXlbQbdPrl0Rve5zK7uO1uHQ/i9K3k3Z3Xa34tHdj > XgKJfgDL2sX+1vm1Rlw6aOLesPfsv5L3WSnWVMia9c4+oWoox7UQ2sO2oq5bTVs0 > ey2F2PGY7eiPHF/Bp2ffcRInH2X2XG+YNY3txl89x5hgeK5WylTnotabZzKaSYJt > /XOT7JdfmyBqxo9fpltX9JGuIEZh9Lck7a8+G+UlkFgwmMx9YRaYZtFDKw/u27BT > CMfUyhwvVXAC7Mgqi+D2Cvu7CBroe7vMp1Y0cnutDoAcFYPVe/143cyV+66EjX9d > PIkmcRGf99pj1yzJpZ8tZJr3O8iQXeDAVsf+imH+8KRWdbwjIZZXpGPlwr8Vts0G > mI0/3twjskrIkpTMPd2XtT9BqmbyrqvidMRLdDSq+sEEyEY1K3vqny0en3oYX8EX > 8RC4MkJN/7rZjZyOsTWRA+J474qUJmK+m2MmLzp25Fr1ivcjd5XcQR77uGSpKjWR > qgTIsl1GY1zR1RLsoRKQ > =4hLW > -----END PGP SIGNATURE----- >
