Question


Anthem Blue Cross Blue Shield
US
Last activity: 21 May 2020 1:53 EDT
Kafka Dataset not able to consume data.
Hello all,
We are using kafka dataset to consume data to our system. Then I am doing test connectivity I am able to connect from the kafka dataset configurations.
But, when external put some files on the kafka topic we were not able to consume records. Dataflow is showing in progress but, I am not able to see any input records. I can not see any errors in logs as well. please help me to resolve this issue.
I have enabled following loggers to debug.
com.pega.dsm.dnode
org.apache.kafka
Thanks.
***Edited by Moderator Marissa to move from Pega Academy to Pega Support Community; update platform capability tags***


Pegasystems Inc.
IN
we are seeing some issue with data-set execution #Any insights on this ...
2020-05-21 05:48:12,503 [ WebContainer : 4] [ STANDARD] [ ] [XXXXXXXX:02.06.01] (XXX_XXXX_XXXX_Data_TrADS.Action) ERROR ah-hhhhhhhhhhhhhhhhhhh.com|XX.XXX.XXX.XX user- [SystemT][PublishToKafka][Error] ** Java Exception: java.lang.RuntimeException: com.pega.dsm.dnode.api.BatchRecordException java.lang.Throwable at com.pegarules.generated.activity.ra_action_publishtokafka_f07b8373cee9ef3a9627099e5703b239.step3_circum0(ra_action_publishtokafka_f07b8373cee9ef3a9627099e5703b239.java:387) ~[?:?] at com.pegarules.generated.activity.ra_action_publishtokafka_f07b8373cee9ef3a9627099e5703b239.perform(ra_action_publishtokafka_f07b8373cee9ef3a9627099e5703b239.java:108) ~[?:?] at com.pega.pegarules.session.internal.mgmt.Executable.doActivity(Executable.java:3597) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.mgmt.Executable.invokeActivity(Executable.java:10845) ~[prprivate.jar:?] at com.pegarules.generated.activity.sh_action_resendtotrads_86e83064ed62dbab22c1f3d36d24f577.step6_circum0(sh_action_resendtotrads_86e83064ed62dbab22c1f3d36d24f577.java:308) ~[?:?] at com.pegarules.generated.activity.sh_action_resendtotrads_86e83064ed62dbab22c1f3d36d24f577.perform(sh_action_resendtotrads_86e83064ed62dbab22c1f3d36d24f577.java:94) ~[?:?] at com.pega.pegarules.session.internal.mgmt.Executable.doActivity(Executable.java:3597) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.mgmt.Executable.doAction(Executable.java:3330) ~[prprivate.jar:?] at com.pegarules.generated.activity.ra_action_pyrunrecordexecute_f7d1e1c1513eef165c7ecbe136f3addd.step2_circum0(ra_action_pyrunrecordexecute_f7d1e1c1513eef165c7ecbe136f3addd.java:249) ~[?:?] at com.pegarules.generated.activity.ra_action_pyrunrecordexecute_f7d1e1c1513eef165c7ecbe136f3addd.perform(ra_action_pyrunrecordexecute_f7d1e1c1513eef165c7ecbe136f3addd.java:88) ~[?:?] at com.pega.pegarules.session.internal.mgmt.Executable.doActivity(Executable.java:3597) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.mgmt.Executable.invokeActivity(Executable.java:10845) ~[prprivate.jar:?] at com.pegarules.generated.activity.ra_action_pzrunrecordexecute_c38281e0bb63f7c4aab88814460c8435.step11_circum0(ra_action_pzrunrecordexecute_c38281e0bb63f7c4aab88814460c8435.java:1027) ~[?:?] at com.pegarules.generated.activity.ra_action_pzrunrecordexecute_c38281e0bb63f7c4aab88814460c8435.perform(ra_action_pzrunrecordexecute_c38281e0bb63f7c4aab88814460c8435.java:246) ~[?:?] at com.pega.pegarules.session.internal.mgmt.Executable.doActivity(Executable.java:3597) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.mgmt.base.ThreadRunner.runActivitiesAlt(ThreadRunner.java:646) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.mgmt.PRThreadImpl.runActivitiesAlt(PRThreadImpl.java:481) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.engineinterface.service.HttpAPI.runActivities(HttpAPI.java:3414) ~[prprivate.jar:?] at com.pega.pegarules.session.external.engineinterface.service.EngineAPI.processRequestInner(EngineAPI.java:399) ~[prenginext.jar:?] at sun.reflect.GeneratedMethodAccessor147.invoke(Unknown Source) ~[?:?] at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:55) ~[?:1.8.0] at java.lang.reflect.Method.invoke(Method.java:508) ~[?:1.8.0] at com.pega.pegarules.session.internal.PRSessionProviderImpl.performTargetActionWithLock(PRSessionProviderImpl.java:1338) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.PRSessionProviderImpl.doWithRequestorLocked(PRSessionProviderImpl.java:1075) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.PRSessionProviderImpl.doWithRequestorLocked(PRSessionProviderImpl.java:930) ~[prprivate.jar:?] at com.pega.pegarules.session.external.engineinterface.service.EngineAPI.processRequest(EngineAPI.java:336) ~[prenginext.jar:?] at com.pega.pegarules.session.internal.engineinterface.service.HttpAPI.invoke(HttpAPI.java:854) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.engineinterface.etier.impl.EngineImpl._invokeEngine_privact(EngineImpl.java:331) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.engineinterface.etier.impl.EngineImpl.invokeEngine(EngineImpl.java:274) ~[prprivate.jar:?] at com.pega.pegarules.session.internal.engineinterface.etier.ejb.EngineBean.invokeEngine(EngineBean.java:225) ~[prprivate.jar:?] at sun.reflect.GeneratedMethodAccessor92.invoke(Unknown Source) ~[?:?] at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:55) ~[?:1.8.0] at java.lang.reflect.Method.invoke(Method.java:508) ~[?:1.8.0] at com.pega.pegarules.internal.bootstrap.PRBootstrap.invokeMethod(PRBootstrap.java:370) ~[prbootstrap-7.3.1-218.jar:7.3.1-218] at com.pega.pegarules.internal.bootstrap.PRBootstrap.invokeMethodPropagatingThrowable(PRBootstrap.java:411) ~[prbootstrap-7.3.1-218.jar:7.3.1-218] at com.pega.pegarules.boot.internal.extbridge.AppServerBridgeToPega.invokeMethodPropagatingThrowable(AppServerBridgeToPega.java:224) ~[prbootstrap-api-7.3.1-218.jar:7.3.1-218] at com.pega.pegarules.boot.internal.extbridge.AppServerBridgeToPega.invokeMethodPropagatingException(AppServerBridgeToPega.java:246) ~[prbootstrap-api-7.3.1-218.jar:7.3.1-218] at com.pega.pegarules.internal.etier.ejb.EngineBeanBoot.invokeEngine(EngineBeanBoot.java:168) ~[prbeans.jar:7.3.1-218] at com.pega.pegarules.internal.etier.interfaces.EJSLocalStatelessEngineBMT_f2439d86.invokeEngine(Unknown Source) ~[prbeans.jar:7.3.1-218] at com.pega.pegarules.priv.context.JNDIEnvironment.invokeEngineInner(JNDIEnvironment.java:278) ~[prpublic.jar:?] at com.pega.pegarules.priv.context.JNDIEnvironment.invokeEngine(JNDIEnvironment.java:223) ~[prpublic.jar:?] at com.pega.pegarules.web.impl.WebStandardImpl.makeEtierRequest(WebStandardImpl.java:678) ~[prwebj2ee.jar:?] at com.pega.pegarules.web.impl.WebStandardImpl.doPost(WebStandardImpl.java:394) ~[prwebj2ee.jar:?] at sun.reflect.GeneratedMethodAccessor143.invoke(Unknown Source) ~[?:?] at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:55) ~[?:1.8.0] at java.lang.reflect.Method.invoke(Method.java:508) ~[?:1.8.0] at com.pega.pegarules.internal.bootstrap.PRBootstrap.invokeMethod(PRBootstrap.java:370) ~[prbootstrap-7.3.1-218.jar:7.3.1-218] at com.pega.pegarules.internal.bootstrap.PRBootstrap.invokeMethodPropagatingThrowable(PRBootstrap.java:411) ~[prbootstrap-7.3.1-218.jar:7.3.1-218] at com.pega.pegarules.boot.internal.extbridge.AppServerBridgeToPega.invokeMethodPropagatingThrowable(AppServerBridgeToPega.java:224) ~[prbootstrap-api-7.3.1-218.jar:7.3.1-218] at com.pega.pegarules.boot.internal.extbridge.AppServerBridgeToPega.invokeMethod(AppServerBridgeToPega.java:273) ~[prbootstrap-api-7.3.1-218.jar:7.3.1-218] at com.pega.pegarules.internal.web.servlet.WebStandardBoot.doPost(WebStandardBoot.java:129) ~[prbootstrap-api-7.3.1-218.jar:7.3.1-218] at com.pega.pegarules.internal.web.servlet.WebStandardBoot.doGet(WebStandardBoot.java:100) ~[prbootstrap-api-7.3.1-218.jar:7.3.1-218] at javax.servlet.http.HttpServlet.service(HttpServlet.java:687) ~[javax.j2ee.servlet.jar:?] at javax.servlet.http.HttpServlet.service(HttpServlet.java:790) ~[javax.j2ee.servlet.jar:?] at com.ibm.ws.webcontainer.servlet.ServletWrapper.service(ServletWrapper.java:1235) ~[com.ibm.ws.webcontainer.jar:?] at com.ibm.ws.webcontainer.servlet.ServletWrapper.handleRequest(ServletWrapper.java:779) ~[com.ibm.ws.webcontainer.jar:?] at com.ibm.ws.webcontainer.servlet.ServletWrapper.handleRequest(ServletWrapper.java:478) ~[com.ibm.ws.webcontainer.jar:?] at com.ibm.ws.webcontainer.servlet.ServletWrapperImpl.handleRequest(ServletWrapperImpl.java:179) ~[com.ibm.ws.webcontainer.jar:?] at com.ibm.ws.webcontainer.filter.WebAppFilterManager.invokeFilters(WebAppFilterManager.java:1124) ~[com.ibm.ws.webcontainer.jar:?] at com.ibm.ws.webcontainer.servlet.CacheServletWrapper.handleRequest(CacheServletWrapper.java:82) ~[com.ibm.ws.webcontainer.jar:?] at com.ibm.ws.webcontainer.WebContainer.handleRequest(WebContainer.java:963) ~[com.ibm.ws.webcontainer.jar:?] at com.ibm.ws.webcontainer.WSWebContainer.handleRequest(WSWebContainer.java:1817) ~[com.ibm.ws.webcontainer.jar:?] at com.ibm.ws.webcontainer.channel.WCChannelLink.ready(WCChannelLink.java:382) ~[com.ibm.ws.webcontainer.jar:?] at com.ibm.ws.http.channel.inbound.impl.HttpInboundLink.handleDiscrimination(HttpInboundLink.java:465) ~[com.ibm.ws.runtime.jar:?] at com.ibm.ws.http.channel.inbound.impl.HttpInboundLink.handleNewRequest(HttpInboundLink.java:532) ~[com.ibm.ws.runtime.jar:?] at com.ibm.ws.http.channel.inbound.impl.HttpInboundLink.processRequest(HttpInboundLink.java:318) ~[com.ibm.ws.runtime.jar:?] at com.ibm.ws.http.channel.inbound.impl.HttpICLReadCallback.complete(HttpICLReadCallback.java:88) ~[com.ibm.ws.runtime.jar:?] at com.ibm.ws.ssl.channel.impl.SSLReadServiceContext$SSLReadCompletedCallback.complete(SSLReadServiceContext.java:1833) ~[com.ibm.ws.runtime.jar:?] at com.ibm.ws.tcp.channel.impl.AioReadCompletionListener.futureCompleted(AioReadCompletionListener.java:175) ~[com.ibm.ws.runtime.jar:?] at com.ibm.io.async.AbstractAsyncFuture.invokeCallback(AbstractAsyncFuture.java:217) ~[com.ibm.ws.runtime.jar:?] at com.ibm.io.async.AsyncChannelFuture$1.run(AsyncChannelFuture.java:205) ~[com.ibm.ws.runtime.jar:?] at com.ibm.ws.util.ThreadPool$Worker.run(ThreadPool.java:1909) ~[com.ibm.ws.runtime.jar:?]
Hi,
What is the status of stream service in stream landing page(Decisioning-->infrastructure-->services--->stream) ?
Meanwhile you may review the below link.
https://community.pega.com/knowledgebase/articles/pega-marketing/configuring-kafka-process-real-time-events
Thank You,