class ReverseLookupComputingCacheEntryListener<K, V> implements ClusteredCacheEntryListener<K, V>
{
    private final Map<NodeID, Set<K>> reverseCacheRepresentation;
    private final Function<V, Set<NodeID>> ownageDeducer;

    public ReverseLookupComputingCacheEntryListener(@Nonnull final Map<NodeID, Set<K>> reverseCacheRepresentation, @Nonnull final Function<V, Set<NodeID>> ownageDeducer) {
        this.reverseCacheRepresentation = reverseCacheRepresentation;
        this.ownageDeducer = ownageDeducer;
    }

    @Override
    public void entryAdded(@Nonnull final K key, @Nullable final V value, @Nonnull final NodeID initiator)
    {
        final Set<NodeID> newNodesWithThisKey = ownageDeducer.apply(value);
        add(key, newNodesWithThisKey);
    }

    private void add(K key, Set<NodeID> newNodesWithThisKey)
    {
        // Step 1 : update existing entries to the updated situation
        final Iterator<Map.Entry<NodeID, Set<K>>> iter = reverseCacheRepresentation.entrySet().iterator();
        while (iter.hasNext()) {
            final Map.Entry<NodeID, Set<K>> existingEntry = iter.next();
            final NodeID existingEntryNodeID = existingEntry.getKey();
            if (newNodesWithThisKey.contains(existingEntryNodeID)) {
                existingEntry.getValue().add(key);
            }
        }
        // Step 2 : add entries for node ids that were not in the reverse lookup before
        for (final NodeID nodeIdForWhichTheValueExists : newNodesWithThisKey) {
            if (!reverseCacheRepresentation.containsKey(nodeIdForWhichTheValueExists)) {
                reverseCacheRepresentation.computeIfAbsent(nodeIdForWhichTheValueExists, k -> new HashSet<>()).add(key);
            }
        }
    }

    @Override
    public void entryRemoved(@Nonnull final K key, @Nullable final V oldValue, @Nonnull final NodeID initiator) {<FILL_FUNCTION_BODY>}

    @Override
    public void entryUpdated(@Nonnull final K key, @Nullable final V oldValue, @Nullable final V newValue, @Nonnull final NodeID nodeID) {
        final Set<NodeID> nodesToAdd = ownageDeducer.apply(newValue);

        // Remove all entries for the key.
        final Iterator<Map.Entry<NodeID, Set<K>>> iter = reverseCacheRepresentation.entrySet().iterator();
        while (iter.hasNext()) {
            final Map.Entry<NodeID, Set<K>> existingEntry = iter.next();
            final NodeID existingEntryNodeID = existingEntry.getKey();
            if (!nodesToAdd.contains(existingEntryNodeID)) {
                existingEntry.getValue().remove(key);
                if (existingEntry.getValue().isEmpty()) {
                    iter.remove();
                }
            }
        }

        // Add entries for only the latest state of the value.
        add(key, nodesToAdd);
    }

    @Override
    public void entryEvicted(@Nonnull final K key, @Nullable final V oldValue, @Nonnull final NodeID nodeID) {
        entryRemoved(key, oldValue, nodeID);
    }

    @Override
    public void mapCleared(@Nonnull final NodeID nodeID) {
        // This is not necessarily correct, as the node owners aren't recalculated using the function. It probably is
        // sufficient though to work with cluster nodes dropping out of the cluster.
        reverseCacheRepresentation.remove(nodeID);
    }

    @Override
    public void mapEvicted(@Nonnull final NodeID nodeID) {
        mapCleared(nodeID);
    }
}

        final Iterator<Map.Entry<NodeID, Set<K>>> iter = reverseCacheRepresentation.entrySet().iterator();
        while (iter.hasNext()) {
            final Map.Entry<NodeID, Set<K>> existingEntry = iter.next();
            existingEntry.getValue().remove(key);
            if (existingEntry.getValue().isEmpty()) {
                iter.remove();
            }
        }
    
<no_super_class>