Skip to main content

This content is closed to future replies and is no longer being maintained or updated.
Links may no longer function. If you have a similar request, please write a new post.

Question

 Vijay Vodnala (VVODNALA)
Centene

Centene
US
VVODNALA Member since 2012 1 post
Centene
Posted: Jan 19, 2021
Last activity: Jul 18, 2023
Posted: 19 Jan 2021 13:22 EST
Last activity: 18 Jul 2023 7:27 EDT
Closed
Solved

Remove items in "Ready to process" from Queue Processor

We have a queue processor 'pyProcessNotification' not running and about 100k items are in the "Ready to Process" state. How can we delete/clear the queue since we don't want to trigger notifications for old items. 

Pega version is 8.3.1

Thank you

***Edited by Moderator Marissa to update Support Case Details***
To see attachments, please log in.
Pega Platform 8.3.1
Case Management
Support Case Exists

Accepted Solution

Posted: 4 years ago
Updated: 3 years ago
Posted: 21 Jan 2021 16:16 EST
Updated: 10 Jun 2021 15:29 EDT
Harish Gunneri (Harish_Gunneri)
PEGA
Senior Principal Engineer, Technical Support
Pegasystems Inc.
US

@VVODNALA

 

You can create an activity with one Java step using the attached code.

 

Ensure to update the TopicName field in the attached code with your QP name in Capital letters for which you want to truncate the ready to process queue.

 

Let me know if that helps!!

To see attachments, please log in.
  • Vijay Vodnala Varun Gupta AMRITA LAHIRI Shailendra Kumar Sankhala Som Venkata Kumar Rallapati and 9 More
Posted: 3 years ago
Posted: 10 Jun 2021 12:02 EDT
Raghu Kanamarlaputi (RaghuK60)
Capgemini

Capgemini
IN

@Harish_Gunneri

Hi Harish,

I am also facing the issue. I have tried with the code which you shared but i am getting issue with the withTopicName method. Let me know is there any other way to clear the Ready to process queues from Queue Processor.

Error Details are below:

The method withTopicName(String) is undefined for the type StreamDataSetBuilder.

Thanks in Advance.

 

Regards,

Raghu Kumar K

To see attachments, please log in.
Posted: 1 year ago
Posted: 18 Jul 2023 7:27 EDT
Vamsi Thatavarthi (Vamsi.Thatavarthi)
ING Bank

ING Bank
NL

Hello @Harish_Gunneri   We are on version, 8.6 and when trying to save the activity with the java code you provided, resulted in the following compilation error

The method withTopicName(String) is undefined for the type StreamDataSetBuilder

Compile failed.

 

To see attachments, please log in.
Posted: 3 years ago
Posted: 10 Jun 2021 8:09 EDT
Pratik Agarwal (pratik)
Cognizant
Architect
Cognizant
GB

@VVODNALAItems queued to Queue Processors are pushed to Kafka, not database table.

In order to clear the queues, you can try removing the partition directories for your Queue Processor, which are created on the app filesystem.  These partitions can be located in the 'kafka-data' directory on each app-node that supports 'Stream' service. For example, kafka-data can be located inside the tomcat installation directory for a Pega app running on tomcat.

Prior to removing the affected partitions from each node separately, ensure the app-servers are stopped to prevent replication between the Stream nodes.  Otherwise the Stream service will repopulate the nodes with partition data soon as you remove it from any node within the cluster.

Let me know how this goes.

Pratik Agarwal

To see attachments, please log in.
  • Vijay Vodnala Pratik Agarwal Ramsurendar Govindarajan
Posted: 3 years ago
Posted: 16 Jun 2021 8:26 EDT
Harish Gunneri (Harish_Gunneri)
PEGA
Senior Principal Engineer, Technical Support
Pegasystems Inc.
US

@RaghuK60 -@RaghuK60@RaghuK60Please@RaghuK60use@RaghuK60the@RaghuK60attached@RaghuK60java@RaghuK60snippet@RaghuK60for@RaghuK60the@RaghuK60latest@RaghuK60versions.

To see attachments, please log in.
  • Venkatesh Gudeenti AMRITA LAHIRI Avinash Haridasu Harjot Kaur Bajwa Venkata Rakhesh Godavarthi and 2 More
Posted: 2 years ago
Updated: 2 years ago
Posted: 13 Apr 2022 16:33 EDT
Updated: 13 Apr 2022 16:34 EDT
Venkata Damarnath Aluru (VenkataDamarnathA)
Optum

Optum
US

Hi For V8.2.8, here is the code  

java.util.ArrayList<String> partitionKeys = new java.util.ArrayList<>();  partitionKeys.add(".pzMessageContent.pzInsKey");  com.pega.dsm.dnode.api.dataset.DataSet dataset = com.pega.dsm.dnode.api.dataset.stream.StreamDataSet.builder() .withClassName("System-Message-QueueProcessor") .withTopicName("<QP_NAME IN CAPITAL LETTER>") .withPartitionKeys(partitionKeys) .withIdempotentSave(true) .withOperation(com.pega.dsm.dnode.api.dataset.operation.SaveOperation.NAME) .withOperation(com.pega.dsm.dnode.impl.dataset.kafka.KafkaGetTopicsOperation.NAME) .withOperation(com.pega.dsm.dnode.api.dataset.operation.BrowseOperation.NAME) .withOperation(com.pega.dsm.dnode.impl.dataset.kafka.KafkaDropOperation.NAME) .withOperation(com.pega.dsm.dnode.impl.dataset.kafka.KafkaTruncateOperation.NAME) .withSerde(com.pega.dsm.kafka.api.serde.ClipboardPageStreamSerde.create()) .build(tools); com.pega.dsm.dnode.api.dataset.operation.Operation operation = dataset.getOperationByName(com.pega.dsm.dnode.impl.dataset.kafka.KafkaTruncateOperation.NAME); ((com.pega.dsm.dnode.impl.dataset.kafka.KafkaTruncateOperation)operation).truncate().await(tools);

 

To see attachments, please log in.
Posted: 2 years ago
Posted: 11 Oct 2022 17:24 EDT
Venkata Ramana Keesara (venkatak5332)
Ai4Process Ltd
Senior Consultant
Ai4Process Ltd
GB

@Harish_Gunneriwhile runnig the activity with the above java code getting this error.

 Java Exception:     java.lang.RuntimeException: java.util.concurrent.ExecutionException: com.pega.fnx.stream.spi.StreamServiceException: Unknown stream ProcessFailedEvents   

Can you please let me know solution for this.

 

To see attachments, please log in.

We'd prefer it if you saw us at our best.

Pega Collaboration Center has detected you are using a browser which may prevent you from experiencing the site as intended. To improve your experience, please update your browser.

Close Deprecation Notice