public void flushPendingChanges( boolean sendToCluster )
    {
        // forward to other cluster members and wait for response
        if (sendToCluster) {
            CacheFactory.doSynchronousClusterTask(new FlushTask(), false);
        }

        // TODO: figure out if it's required to first flush pending nodes, cluster-wide, synchronously, before flushing items.
        flushPendingNodes();

        if (itemsToAdd.isEmpty() && itemsToDelete.isEmpty()) {
            return;	 // Nothing left to do for this cluster member.
        }

        List<PublishedItem> addList;
        List<PublishedItem> delList;

        // Swap pending items so we can parse and save the contents from this point in time
        // while not blocking new entries from being cached.
        synchronized(itemsPending)
        {
            addList = new ArrayList<>( itemsToAdd );
            delList = new ArrayList<>( itemsToDelete );

            itemsToAdd = new ConcurrentLinkedDeque<>();
            itemsToDelete = new ConcurrentLinkedDeque<>();

            // Ensure pending items are available via the item read cache;
            // this allows the item(s) to be fetched by other request threads
            // while being written to the DB from this thread
            // TODO Determine if this works as intended when items are being queued for adding as well as removal.
            int copied = 0;
            for ( PublishedItem.UniqueIdentifier key : itemsPending.keySet()) {
                if (!itemCache.containsKey(key)) {
                    itemCache.put(key, itemsPending.get(key));
                    copied++;
                }
            }
            if (log.isDebugEnabled() && copied > 0) {
                log.debug("Added " + copied + " pending items to published item cache");
            }
            itemsPending.clear();
        }

        delegate.bulkPublishedItems( addList, delList );
        flushPendingChanges( sendToCluster );
    }
Persists any changes that have been applied to the caches by invoking the relevant methods of the delegate.

A flush can be performed either local (typically used to periodically persist data) or cluster-wide (useful to
ensure that a particular state is reached, cluster-wide).

@param sendToCluster set to 'true' to trigger a flush on all cluster nodes. If false, a flush will occur only on
                     the local node.
xmppserver/src/main/java/org/jivesoftware/openfire/pubsub/CachingPubsubPersistenceProvider.java