modules/andes-core/broker/src/main/java/org/wso2/andes/store/rdbms/RDBMSMessageStoreImpl.java
@Override
    public void moveMetadataToDLC(List<AndesMessageMetadata> messages, String dlcQueueName) throws AndesException {
        Connection connection = null;
        PreparedStatement metadataPS = null;
        PreparedStatement expiryDataPS = null;

        Context moveMetadataToDLCContext = MetricManager.timer(MetricsConstants.MOVE_METADATA_TO_DLC, Level.INFO)
                .start();
        Context contextWrite = MetricManager.timer(MetricsConstants.DB_WRITE, Level.INFO).start();
        LongArrayList messageIDsToRemoveFromCache = new LongArrayList();

        try {
            connection = getConnection();
            metadataPS = connection.prepareStatement(RDBMSConstants.PS_MOVE_METADATA_TO_DLC);
            int cachedQueueId = getCachedQueueID(dlcQueueName);
            expiryDataPS = connection.prepareStatement(RDBMSConstants.PS_UPDATE_DLC_STATUS_IN_EXPIRY_TABLE);
            for (AndesMessageMetadata message : messages) {
                messageIDsToRemoveFromCache.add(message.getMessageID());
                metadataPS.setInt(1, cachedQueueId);
                metadataPS.setLong(2, message.getMessageID());
                expiryDataPS.setInt(1, cachedQueueId);
                expiryDataPS.setLong(2, message.getMessageID());
                metadataPS.addBatch();
                expiryDataPS.addBatch();
            }
            metadataPS.executeBatch();
            expiryDataPS.executeBatch();
            //remove messages from cache
            removeFromCache(messageIDsToRemoveFromCache);
            connection.commit();
        } catch (SQLException e) {
            rollback(connection, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            throw rdbmsStoreUtils.convertSQLException("Error occurred while moving message metadata to dead letter "
                    + "channel.", e);
        } finally {
            contextWrite.stop();
            moveMetadataToDLCContext.stop();
            close(connection, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            close(metadataPS, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            close(expiryDataPS, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
        }
    }
/**
     * Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages
     * {@inheritDoc}
     */
Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages
Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages
@Override
    public void moveMetadataToDLC(List<AndesMessageMetadata> messages, String dlcQueueName) throws AndesException {
        Connection connection = null;
        PreparedStatement metadataPS = null;
        PreparedStatement expiryDataPS = null;
        Context moveMetadataToDLCContext = MetricManager.timer(MetricsConstants.MOVE_METADATA_TO_DLC, Level.INFO)
                .start();
        Context contextWrite = MetricManager.timer(MetricsConstants.DB_WRITE, Level.INFO).start();
        LongArrayList messageIDsToRemoveFromCache = new LongArrayList();
        try {
            connection = getConnection();
            metadataPS = connection.prepareStatement(RDBMSConstants.PS_MOVE_METADATA_TO_DLC);
            int cachedQueueId = getCachedQueueID(dlcQueueName);
            expiryDataPS = connection.prepareStatement(RDBMSConstants.PS_UPDATE_DLC_STATUS_IN_EXPIRY_TABLE);
            for (AndesMessageMetadata message : messages) {
                messageIDsToRemoveFromCache.add(message.getMessageID());
                metadataPS.setInt(1, cachedQueueId);
                metadataPS.setLong(2, message.getMessageID());
                expiryDataPS.setInt(1, cachedQueueId);
                expiryDataPS.setLong(2, message.getMessageID());
                metadataPS.addBatch();
                expiryDataPS.addBatch();
            }
            metadataPS.executeBatch();
            expiryDataPS.executeBatch();
            removeFromCache(messageIDsToRemoveFromCache);
            connection.commit();
        } catch (SQLException e) {
            rollback(connection, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            throw rdbmsStoreUtils.convertSQLException("Error occurred while moving message metadata to dead letter "
                    + "channel.", e);
        } finally {
            contextWrite.stop();
            moveMetadataToDLCContext.stop();
            close(connection, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            close(metadataPS, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            close(expiryDataPS, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
        }
    }
@ Override public void moveMetadataToDLC ( List < AndesMessageMetadata > messages , String dlcQueueName ) throws AndesException { Connection connection = null ; PreparedStatement metadataPS = null ; PreparedStatement expiryDataPS = null ; Context moveMetadataToDLCContext = MetricManager . timer ( MetricsConstants . MOVE_METADATA_TO_DLC , Level . INFO ) . start ( ) ; Context contextWrite = MetricManager . timer ( MetricsConstants . DB_WRITE , Level . INFO ) . start ( ) ; LongArrayList messageIDsToRemoveFromCache = new LongArrayList ( ) ; try { connection = getConnection ( ) ; metadataPS = connection . prepareStatement ( RDBMSConstants . PS_MOVE_METADATA_TO_DLC ) ; int cachedQueueId = getCachedQueueID ( dlcQueueName ) ; expiryDataPS = connection . prepareStatement ( RDBMSConstants . PS_UPDATE_DLC_STATUS_IN_EXPIRY_TABLE ) ; for ( AndesMessageMetadata message : messages ) { messageIDsToRemoveFromCache . add ( message . getMessageID ( ) ) ; metadataPS . setInt ( 1 , cachedQueueId ) ; metadataPS . setLong ( 2 , message . getMessageID ( ) ) ; expiryDataPS . setInt ( 1 , cachedQueueId ) ; expiryDataPS . setLong ( 2 , message . getMessageID ( ) ) ; metadataPS . addBatch ( ) ; expiryDataPS . addBatch ( ) ; } metadataPS . executeBatch ( ) ; expiryDataPS . executeBatch ( ) ; removeFromCache ( messageIDsToRemoveFromCache ) ; connection . commit ( ) ; } catch ( SQLException e ) { rollback ( connection , RDBMSConstants . TASK_MOVING_METADATA_TO_DLC ) ; throw rdbmsStoreUtils . convertSQLException ( "Error occurred while moving message metadata to dead letter " + "channel." , e ) ; } finally { contextWrite . stop ( ) ; moveMetadataToDLCContext . stop ( ) ; close ( connection , RDBMSConstants . TASK_MOVING_METADATA_TO_DLC ) ; close ( metadataPS , RDBMSConstants . TASK_MOVING_METADATA_TO_DLC ) ; close ( expiryDataPS , RDBMSConstants . TASK_MOVING_METADATA_TO_DLC ) ; } }
Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages
 {@inheritDoc}
Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages { @inheritDoc }
//remove messages from cache