@Override
    public void run() {
        while (run) {
            QueuedCudStreamProcessor queuedCudStreamProcessor = null;
            try {
                queuedCudStreamProcessor = this.cudStreamProcessorQueue.take();
            } catch (InterruptedException e) {
                log.warn("Thread interrupted. Error when trying to retrieve queued values." + e.getMessage());
            }
            if (null != queuedCudStreamProcessor) {
                if (log.isDebugEnabled()) {
                    log.debug("Current queue size is = " + cudStreamProcessorQueue.size());
                }
                int i = 0;
                while (true) {
                    i++;
                    try {
                        ComplexEventChunk complexEventChunk = new ComplexEventChunk();
                        complexEventChunk.add(queuedCudStreamProcessor.getStreamEvent());
                        if (log.isDebugEnabled()) {
                            log.debug("Starting processing for duration " + queuedCudStreamProcessor.getDuration());
                        }
                        queuedCudStreamProcessor.getCudStreamProcessor().
                                process(complexEventChunk);
                        complexEventChunk.clear();
                        if (log.isDebugEnabled()) {
                            log.debug("End processing for duration " + queuedCudStreamProcessor.getDuration());
                        }
                        break;
                    } catch (Exception e) {
                        if (e.getCause() instanceof SQLException) {
                            if (e.getCause().getLocalizedMessage().contains("try restarting transaction") && i < 3) {
                                log.error("Error occurred while executing the aggregation for data between " +
                                        queuedCudStreamProcessor.getStartTimeOfNewAggregates() + " - " +
                                        queuedCudStreamProcessor.getEmittedTime() + " for duration " +
                                        queuedCudStreamProcessor.getDuration() +
                                        " Retrying the transaction attempt " + (i - 1), e);
                                try {
                                    Thread.sleep(3000);
                                } catch (InterruptedException interruptedException) {
                                    log.error("Thread sleep interrupted while waiting to re-execute the " +
                                            "aggregation query for duration " +
                                            queuedCudStreamProcessor.getDuration(), interruptedException);
                                }
                                continue;
                            }
                            log.error("Error occurred while executing the aggregation for data between "
                                    + queuedCudStreamProcessor.getStartTimeOfNewAggregates() + " - " +
                                    queuedCudStreamProcessor.getEmittedTime() + " for duration " +
                                    queuedCudStreamProcessor.getDuration() +
                                    ". Attempted re-executing the query for 9 seconds. " +
                                    "This Should be investigated since this will lead to a data mismatch\n", e);
                        } else {
                            log.error("Error occurred while executing the aggregation for data between "
                                    + queuedCudStreamProcessor.getStartTimeOfNewAggregates() + " - " +
                                    queuedCudStreamProcessor.getEmittedTime() + " for duration \n" +
                                    queuedCudStreamProcessor.getDuration(), e);
                        }
                        break;
                    }
                }
            }
        }
    }
Implements functions to manage the queue
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/persistedaggregation/CudStreamProcessorQueueManager.java