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

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

        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)
        {
            // find items for node.

            // Split the to-do list in two parts: one that contains items for the node of interest, and the rest.
            final Map<Boolean, List<PublishedItem>> partsToAdd = itemsToAdd.stream().collect(
                    Collectors.partitioningBy( publishedItem -> nodeUniqueId.equals( publishedItem.getNode().getUniqueIdentifier() ) )
            );
            addList = new ArrayList<>( partsToAdd.get( true ) ); // All elements that match must be processed.
            itemsToAdd.retainAll( partsToAdd.get(false) ); // Non-matching elements remain on the to-do list.

            // Split the to-do list in two parts: one that contains items for the node of interest, and the rest.
            final Map<Boolean, List<PublishedItem>> partsToDelete = itemsToDelete.stream().collect(
                    Collectors.partitioningBy( publishedItem -> nodeUniqueId.equals( publishedItem.getNode().getUniqueIdentifier() ) )
            );
            delList = new ArrayList<>( partsToDelete.get( true ) ); // All elements that match must be processed.
            itemsToDelete.retainAll( partsToDelete.get(false) ); // Non-matching elements remain on the to-do list.

            // 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 (final PublishedItem itemToAdd : itemsToAdd) {
                final PublishedItem.UniqueIdentifier key = itemToAdd.getUniqueIdentifier();
                if (!itemCache.containsKey(key)) {
                    itemsPending.remove( key );
                    itemCache.put(key, itemToAdd);
                    copied++;
                }
            }
            if (log.isDebugEnabled() && copied > 0) {
                log.debug("Added " + copied + " pending items to published item cache");
            }
        }

        delegate.bulkPublishedItems( addList, delList );
    }
Cache for recently accessed published items.
xmppserver/src/main/java/org/jivesoftware/openfire/pubsub/CachingPubsubPersistenceProvider.java