modules/andes-core/broker/src/main/java/org/wso2/andes/store/rdbms/RDBMSMessageStoreImpl.java
@Override
    public void storeRetainedMessages(Map<String, AndesMessage> retainMap) throws AndesException {

        Connection connection = null;

        PreparedStatement updateMetadataPreparedStatement = null;
        PreparedStatement deleteContentPreparedStatement = null;
        PreparedStatement deleteMetadataPreparedStatement = null;
        PreparedStatement insertContentPreparedStatement = null;
        Context contextWrite = MetricManager.timer(MetricsConstants.DB_WRITE, Level.INFO).start();

        boolean batchEmpty = true;

        try {
            connection = getConnection();

            updateMetadataPreparedStatement = connection.prepareStatement(RDBMSConstants.PS_UPDATE_RETAINED_METADATA);
            deleteContentPreparedStatement = connection.prepareStatement(RDBMSConstants.PS_DELETE_RETAIN_MESSAGE_PARTS);
            deleteMetadataPreparedStatement = connection
                    .prepareStatement(RDBMSConstants.PS_DELETE_RETAIN_MESSAGE_METADATA);
            insertContentPreparedStatement = connection.prepareStatement(RDBMSConstants.PS_INSERT_RETAIN_MESSAGE_PART);
            for (AndesMessage message : retainMap.values()) {

                AndesMessageMetadata metadata = message.getMetadata();
                String destination = metadata.getDestination();
                RetainedItemData retainedItemData = getRetainedTopicID(connection, destination);

                if (null != retainedItemData) {

                    if (batchEmpty) {
                        batchEmpty = false;
                    }

                    addRetainedMessageToUpdateBatch(updateMetadataPreparedStatement, deleteContentPreparedStatement,
                            deleteMetadataPreparedStatement, insertContentPreparedStatement, message, metadata,
                            retainedItemData);
                    retainedItemData.messageID = metadata.getMessageID();

                } else {
                    // Retain message shouldn't create a retain entry if it receives an empty payload.
                    if (!message.getContentChunkList().isEmpty()) {
                        // if first chunk doesn't contain any data that payload is empty. Therefore
                        // no need to store the retain message.
                        if (message.getContentChunkList().get(0).getDataLength() != 0) {
                            createRetainedEntry(connection, message);
                        }
                    }
                }

            }

            if (!batchEmpty) {
                deleteContentPreparedStatement.executeBatch();
                deleteMetadataPreparedStatement.executeBatch();
                updateMetadataPreparedStatement.executeBatch();
                insertContentPreparedStatement.executeBatch();
                connection.commit();
            }

        } catch (SQLException e) {
            rollback(connection, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            throw rdbmsStoreUtils.convertSQLException("Error occurred while adding retained message content to DB ", e);

        } finally {
            contextWrite.stop();
            close(updateMetadataPreparedStatement, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(deleteContentPreparedStatement, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(deleteMetadataPreparedStatement, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(insertContentPreparedStatement, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(connection, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
        }
    }
/**
     * Store retained messages in RDBMS message stores
     * <p>
     * {@inheritDoc}
     */
Store retained messages in RDBMS message stores

Store retained messages in RDBMS message stores
@Override
    public void storeRetainedMessages(Map<String, AndesMessage> retainMap) throws AndesException {
        Connection connection = null;
        PreparedStatement updateMetadataPreparedStatement = null;
        PreparedStatement deleteContentPreparedStatement = null;
        PreparedStatement deleteMetadataPreparedStatement = null;
        PreparedStatement insertContentPreparedStatement = null;
        Context contextWrite = MetricManager.timer(MetricsConstants.DB_WRITE, Level.INFO).start();
        boolean batchEmpty = true;
        try {
            connection = getConnection();
            updateMetadataPreparedStatement = connection.prepareStatement(RDBMSConstants.PS_UPDATE_RETAINED_METADATA);
            deleteContentPreparedStatement = connection.prepareStatement(RDBMSConstants.PS_DELETE_RETAIN_MESSAGE_PARTS);
            deleteMetadataPreparedStatement = connection
                    .prepareStatement(RDBMSConstants.PS_DELETE_RETAIN_MESSAGE_METADATA);
            insertContentPreparedStatement = connection.prepareStatement(RDBMSConstants.PS_INSERT_RETAIN_MESSAGE_PART);
            for (AndesMessage message : retainMap.values()) {
                AndesMessageMetadata metadata = message.getMetadata();
                String destination = metadata.getDestination();
                RetainedItemData retainedItemData = getRetainedTopicID(connection, destination);
                if (null != retainedItemData) {
                    if (batchEmpty) {
                        batchEmpty = false;
                    }
                    addRetainedMessageToUpdateBatch(updateMetadataPreparedStatement, deleteContentPreparedStatement,
                            deleteMetadataPreparedStatement, insertContentPreparedStatement, message, metadata,
                            retainedItemData);
                    retainedItemData.messageID = metadata.getMessageID();
                } else {
                    if (!message.getContentChunkList().isEmpty()) {
                        if (message.getContentChunkList().get(0).getDataLength() != 0) {
                            createRetainedEntry(connection, message);
                        }
                    }
                }
            }
            if (!batchEmpty) {
                deleteContentPreparedStatement.executeBatch();
                deleteMetadataPreparedStatement.executeBatch();
                updateMetadataPreparedStatement.executeBatch();
                insertContentPreparedStatement.executeBatch();
                connection.commit();
            }
        } catch (SQLException e) {
            rollback(connection, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            throw rdbmsStoreUtils.convertSQLException("Error occurred while adding retained message content to DB ", e);
        } finally {
            contextWrite.stop();
            close(updateMetadataPreparedStatement, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(deleteContentPreparedStatement, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(deleteMetadataPreparedStatement, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(insertContentPreparedStatement, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(connection, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
        }
    }
@ Override public void storeRetainedMessages ( Map < String , AndesMessage > retainMap ) throws AndesException { Connection connection = null ; PreparedStatement updateMetadataPreparedStatement = null ; PreparedStatement deleteContentPreparedStatement = null ; PreparedStatement deleteMetadataPreparedStatement = null ; PreparedStatement insertContentPreparedStatement = null ; Context contextWrite = MetricManager . timer ( MetricsConstants . DB_WRITE , Level . INFO ) . start ( ) ; boolean batchEmpty = true ; try { connection = getConnection ( ) ; updateMetadataPreparedStatement = connection . prepareStatement ( RDBMSConstants . PS_UPDATE_RETAINED_METADATA ) ; deleteContentPreparedStatement = connection . prepareStatement ( RDBMSConstants . PS_DELETE_RETAIN_MESSAGE_PARTS ) ; deleteMetadataPreparedStatement = connection . prepareStatement ( RDBMSConstants . PS_DELETE_RETAIN_MESSAGE_METADATA ) ; insertContentPreparedStatement = connection . prepareStatement ( RDBMSConstants . PS_INSERT_RETAIN_MESSAGE_PART ) ; for ( AndesMessage message : retainMap . values ( ) ) { AndesMessageMetadata metadata = message . getMetadata ( ) ; String destination = metadata . getDestination ( ) ; RetainedItemData retainedItemData = getRetainedTopicID ( connection , destination ) ; if ( null != retainedItemData ) { if ( batchEmpty ) { batchEmpty = false ; } addRetainedMessageToUpdateBatch ( updateMetadataPreparedStatement , deleteContentPreparedStatement , deleteMetadataPreparedStatement , insertContentPreparedStatement , message , metadata , retainedItemData ) ; retainedItemData . messageID = metadata . getMessageID ( ) ; } else { if ( ! message . getContentChunkList ( ) . isEmpty ( ) ) { if ( message . getContentChunkList ( ) . get ( 0 ) . getDataLength ( ) != 0 ) { createRetainedEntry ( connection , message ) ; } } } } if ( ! batchEmpty ) { deleteContentPreparedStatement . executeBatch ( ) ; deleteMetadataPreparedStatement . executeBatch ( ) ; updateMetadataPreparedStatement . executeBatch ( ) ; insertContentPreparedStatement . executeBatch ( ) ; connection . commit ( ) ; } } catch ( SQLException e ) { rollback ( connection , RDBMSConstants . TASK_STORING_RETAINED_MESSAGE ) ; throw rdbmsStoreUtils . convertSQLException ( "Error occurred while adding retained message content to DB " , e ) ; } finally { contextWrite . stop ( ) ; close ( updateMetadataPreparedStatement , RDBMSConstants . TASK_STORING_RETAINED_MESSAGE ) ; close ( deleteContentPreparedStatement , RDBMSConstants . TASK_STORING_RETAINED_MESSAGE ) ; close ( deleteMetadataPreparedStatement , RDBMSConstants . TASK_STORING_RETAINED_MESSAGE ) ; close ( insertContentPreparedStatement , RDBMSConstants . TASK_STORING_RETAINED_MESSAGE ) ; close ( connection , RDBMSConstants . TASK_STORING_RETAINED_MESSAGE ) ; } }
Store retained messages in RDBMS message stores
 <p>
 {@inheritDoc}
Store retained messages in RDBMS message stores <p > { @inheritDoc }
// Retain message shouldn't create a retain entry if it receives an empty payload. // if first chunk doesn't contain any data that payload is empty. Therefore // no need to store the retain message.