Kafka Data Flow Error

chaus5 · June 28, 2023, 4:11pm

Hi All,

Pega Version : 8.7.2

Server : Websphere

DB : Oracle

Established Kerberos connection successfully. Running the dataset manually works fine and able to see the messages.

However , when running the Real Time Data Flow, it gets stuck in ‘Initializing’ with below error:

Failed To Prepare Partitions: org.apache.kafka.common.errors.TimeoutException: Topic <> not present in metadata after 30000 ms.

Also, when we tried to manually run the data set from activity, it fails stating Exception executing the dataset. Please note, we are able to see the jsonData for the events in tracer though.

Error trace when running data set manually from activity:
2023-06-28 04:25:29,160 [ WebContainer : 36] [ STANDARD] (set.kafka.KafkaBrowseOperation) ERROR - Error deserializing stream while reading from Kafka topic com.pega.dsm.dnode.api.dataset.kafka.KafkaDataSet$KafkaDataSetBuilder$$Lambda$4850/0x00000000a4bb44f0@30640d06
at com.pega.dsm.kafka.api.serde.JsonSerde.deserialize(JsonSerde.java:59) ~[d-node.jar:?]
at com.pega.dsm.kafka.impl.serde.ConfiguredPegaSerde.deserialize(ConfiguredPegaSerde.java:59) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.deserializeMessage(KafkaBrowseOperation.java:347) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.convertRecordToClipboardPage(KafkaBrowseOperation.java:302) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:210) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:167) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:157) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecordsAndComplete(KafkaBrowseOperation.java:149) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.lambda$browse$0(KafkaBrowseOperation.java:145) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.reportError(KafkaBrowseOperation.java:261) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:217) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:167) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:157) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecordsAndComplete(KafkaBrowseOperation.java:149) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.lambda$browse$0(KafkaBrowseOperation.java:145) ~[d-node.jar:?]
at com.pega.dsm.kafka.api.serde.JsonSerde.deserialize(JsonSerde.java:59) ~[d-node.jar:?]
at com.pega.dsm.kafka.impl.serde.ConfiguredPegaSerde.deserialize(ConfiguredPegaSerde.java:59) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.deserializeMessage(KafkaBrowseOperation.java:347) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.convertRecordToClipboardPage(KafkaBrowseOperation.java:302) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:210) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:167) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:157) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecordsAndComplete(KafkaBrowseOperation.java:149) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.lambda$browse$0(KafkaBrowseOperation.java:145) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.reportError(KafkaBrowseOperation.java:261) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:217) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:167) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:157) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecordsAndComplete(KafkaBrowseOperation.java:149) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.lambda$browse$0(KafkaBrowseOperation.java:145) ~[d-node.jar:?]
at com.pega.dsm.kafka.api.serde.JsonSerde.deserialize(JsonSerde.java:59) ~[d-node.jar:?]
at com.pega.dsm.kafka.impl.serde.ConfiguredPegaSerde.deserialize(ConfiguredPegaSerde.java:59) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.deserializeMessage(KafkaBrowseOperation.java:347) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.convertRecordToClipboardPage(KafkaBrowseOperation.java:302) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:210) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:167) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:157) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecordsAndComplete(KafkaBrowseOperation.java:149) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.lambda$browse$0(KafkaBrowseOperation.java:145) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.reportError(KafkaBrowseOperation.java:261) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:217) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:167) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:157) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecordsAndComplete(KafkaBrowseOperation.java:149) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.lambda$browse$0(KafkaBrowseOperation.java:145) ~[d-node.jar:?]
at com.pega.dsm.kafka.api.serde.JsonSerde.deserialize(JsonSerde.java:59) ~[d-node.jar:?]
at com.pega.dsm.kafka.impl.serde.ConfiguredPegaSerde.deserialize(ConfiguredPegaSerde.java:59) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.deserializeMessage(KafkaBrowseOperation.java:347) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.convertRecordToClipboardPage(KafkaBrowseOperation.java:302) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:210) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:167) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecords(KafkaBrowseOperation.java:157) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.processRecordsAndComplete(KafkaBrowseOperation.java:149) ~[d-node.jar:?]
at com.pega.dsm.dnode.impl.dataset.kafka.KafkaBrowseOperation.lambda$browse$0(KafkaBrowseOperation.java:145) ~[d-node.jar:?]

Any help would be appreciated.

Anoop_Krishna · June 28, 2023, 4:53pm

@SubhashC3897 we had faced this issue. We raised a SR and Pegasystems externalized kafka and this started working fine.

But in dev initially for testing purpose what we did is,

First created a real time data flow(this went to partition error). → Initializing stage
Then keeping the above one, created another real time data flow instance and this went to in progress state and stopped and removed the above one.

Kiran_Dokka · June 29, 2023, 4:04am

@Anoop Krishna , thank you for your prompt response.

Can you explain about what externalizing KAFKA means?

Fyi, we are the consumers.

Anoop_Krishna · June 29, 2023, 5:08am

@KiranD01 sure, earlier kafka services was embedded in the same server. Later on, as it is pega cloud, pegasystems offloaded kafka services into another server and trafficking would be less and with more performance. This is externaliztion of kafka services

Please check the below URL

https://support.pega.com/question/externalisation-kafka-service

SiddhantJivane · June 29, 2023, 5:48am

@SubhashC3897 ,

Seems like partition related issue,

Try server restart by deleting ‘Kafka-data’ folder from the server.

and run below queries before restart:

truncate table pegadata.pr_data_stream_node_updates;

truncate table pegadata.pr_data_stream_nodes;

truncate table pegadata.pr_data_stream_sessions;

Let me know if this does not work.

Thanks,

Siddhant Suryakant Jivane

chaus5 · July 3, 2023, 10:09am

@SiddhantJivane Thanks for the steps. We did try but the same issues occurs.

SiddhantJivane · July 6, 2023, 1:32am

@SubhashC3897 , Better to raise an SR.

MarijeSchillern · January 26, 2024, 5:22pm

@SubhashC3897 if you did raise an INC support ticket, please provide the id here so that we can help track it.

MarijeSchillern · February 9, 2024, 11:28am

@SubhashC3897 I’ve checked and it seems that you still did not log an INC on the MSP.

If your issue is resolved, please add a reply and mark as Accept Solution so that we can close this thread.

Conversation		Replies	Views
Real time processing data flow failure General lead-system-architect , decision-management , communications-and-media , 8-6	2	822	December 14, 2021
Exception handling for pega Kafka producer and consumer General pega-platform , data-integration , 8-8-5	1	329	March 17, 2025
Use Kafka, Data Set, and Data Flow in Pega Knowledge Share sales-consultant , sales-manager , senior-system-architect , solutions-consultant , solutions-consulting-manager , solutions-engineer , system-architect , lead-system-architect , decisioning-architect , team-lead , low-code-app-development , data-integration , decision-management , enterprise-application-development , business-simulations , data-management , data-model , financial-services , manufacturing , government , insurance , cross-industry , other-industry , consumer-services , communications-and-media , healthcare-and-life-sciences , all-products , pega-infinity , dev-designer-studio	3	1849	May 13, 2025
Kafka Dataset fails when running in Dataflow User Experience user-experience , system-administration , decision-management , data-management	2	161	June 4, 2024
Setting up Kafka and Integrating it with Pega using Stream Dataset , Event strategy and Dataflow General pega-platform , lead-system-architect , case-management , data-integration , decision-management , 8-7-3	4	3239	September 19, 2023

Kafka Data Flow Error

Related topics