class DistributedBatchQueue extends JobQueueCommon {

    @Override
    public void assertHealthy(UUID aggregatorID) {
        try (final Session session = this.factory.openSession()) {
            try {
                OffsetDateTime stuckSince = OffsetDateTime.now(ZoneId.systemDefault()).minusMinutes(3);

                logger.debug(String.format("Checking aggregatorID(%s) for stuck jobs since (%s)...", aggregatorID, stuckSince.toString()));
                Long stuckBatchCount = (Long) session
                        .createQuery("select count(*) from job_queue_batch where aggregatorID = :aggregatorID and status = 1 and updateTime < :updateTime")
                        .setParameter("aggregatorID", aggregatorID)
                        .setParameter("updateTime", stuckSince)
                        .uniqueResult();

                logger.debug(String.format("Found (%d) stuck jobs on aggregatorID(%s).", stuckBatchCount, aggregatorID));

                if (stuckBatchCount > 0) {
                    throw new JobQueueUnhealthy(JOB_UNHEALTHY);
                }
            } catch (JobQueueUnhealthy e) {
                // Rethrow
                throw e;
            } catch (Exception e) {
                throw new JobQueueUnhealthy(DB_UNHEALTHY, e);
            }
        }
    }

    @Inject public  DistributedBatchQueue(
            DPCQueueManagedSessionFactory factory,
            @QueueBatchSize int batchSize,
            MetricRegistry metricRegistry
    );

}