@Test public void ackQueueMessageNotFound() throws Exception { Injector injector = getInjector(); injector.getInstance( App.class ); QueueManager queueManager = injector.getInstance( QueueManager.class ); ActorSystemFig actorSystemFig = injector.getInstance( ActorSystemFig.class ); String region = actorSystemFig.getRegionLocal(); String queueName = "qat_queue_" + RandomStringUtils.randomAlphanumeric( 10 ); queueManager.createQueue( new Queue( queueName ) ); try { UUID queueMessageId = QakkaUtils.getTimeUuid(); QueueActorHelper helper = injector.getInstance( QueueActorHelper.class ); Assert.assertEquals( DistributedQueueService.Status.NOT_INFLIGHT, helper.ackQueueMessage( queueName, queueMessageId ) ); DistributedQueueService distributedQueueService = injector.getInstance( DistributedQueueService.class ); distributedQueueService.shutdown(); } finally { queueManager.deleteQueue( queueName ); } }
DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } }
QueueActorHelper { DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } } }
QueueActorHelper { DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            ); }
QueueActorHelper { DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            );  }
QueueActorHelper { DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            );   }