@Override
    public V compute(K key, BiFunction<K, V, V> recomputeFunction) {
        checkState(!destroyed, destroyedMessage);
        checkNotNull(key, ERROR_NULL_KEY);
        checkNotNull(recomputeFunction, "Recompute function cannot be null");

        AtomicBoolean updated = new AtomicBoolean(false);
        AtomicReference<MapValue<V>> previousValue = new AtomicReference<>();
        MapValue<V> computedValue = items.compute(serializer.copy(key), (k, mv) -> {
            previousValue.set(mv);
            V newRawValue = recomputeFunction.apply(key, mv == null ? null : mv.get());
            if (mv != null && Objects.equals(newRawValue, mv.get())) {
                // value was not updated
                return mv;
            }
            MapValue<V> newValue = new MapValue<>(newRawValue, timestampProvider.apply(key, newRawValue));
            if (mv == null || newValue.isNewerThan(mv)) {
                updated.set(true);
                // We return a copy to ensure updates to peers can be serialized.
                // This prevents replica divergence due to serialization failures.
                return serializer.copy(newValue);
            } else {
                return mv;
            }
        });
        if (updated.get()) {
            notifyPeers(new UpdateEntry<>(key, computedValue), peerUpdateFunction.apply(key, computedValue.get()));
            EventuallyConsistentMapEvent.Type updateType = computedValue.isTombstone() ? REMOVE : PUT;
            V value = computedValue.isTombstone()
                    ? previousValue.get() == null ? null : previousValue.get().get()
                    : computedValue.get();
            if (value != null) {
                notifyListeners(new EventuallyConsistentMapEvent<>(mapName, updateType, key, value));
            }
        }
        return computedValue.get();
    }
@Test
    public void testCompute() throws Exception {
        // Set up expectations of external events to be sent to listeners during
        // the test. These don't use timestamps so we can set them all up at once.
        EventuallyConsistentMapListener<String, String> listener
                = getListener();
        listener.event(new EventuallyConsistentMapEvent<>(
                MAP_NAME, EventuallyConsistentMapEvent.Type.PUT, KEY1, VALUE1));
        listener.event(new EventuallyConsistentMapEvent<>(
                MAP_NAME, EventuallyConsistentMapEvent.Type.REMOVE, KEY1, VALUE1));
        listener.event(new EventuallyConsistentMapEvent<>(
                MAP_NAME, EventuallyConsistentMapEvent.Type.PUT, KEY2, VALUE2));
        replay(listener);

        ecMap.addListener(listener);

        // Put in an initial value
        expectPeerMessage(clusterCommunicator);
        ecMap.compute(KEY1, (k, v) -> VALUE1);
        assertEquals(VALUE1, ecMap.get(KEY1));

        // Remove the value and check the correct internal cluster messages
        // are sent
        expectSpecificMulticastMessage(generateRemoveMessage(KEY1, clockService.peekAtNextTimestamp()),
                UPDATE_MESSAGE_SUBJECT, clusterCommunicator);

        ecMap.compute(KEY1, (k, v) -> null);
        assertNull(ecMap.get(KEY1));

        verify(clusterCommunicator);

        // Remove the same value again. Even though the value is no longer in
        // the map, we expect that the tombstone is updated and another remove
        // event is sent to the cluster and external listeners.
        expectSpecificMulticastMessage(generateRemoveMessage(KEY1, clockService.peekAtNextTimestamp()),
                UPDATE_MESSAGE_SUBJECT, clusterCommunicator);

        ecMap.compute(KEY1, (k, v) -> null);
        assertNull(ecMap.get(KEY1));

        verify(clusterCommunicator);

        // Put in a new value for us to try and remove
        expectPeerMessage(clusterCommunicator);

        ecMap.compute(KEY2, (k, v) -> VALUE2);

        clockService.turnBackTime();

        // Remove should have no effect, since it has an older timestamp than
        // the put. Expect no notifications to be sent out
        reset(clusterCommunicator);
        replay(clusterCommunicator);

        ecMap.compute(KEY2, (k, v) -> null);

        verify(clusterCommunicator);

        // Check that our listener received the correct events during the test
        verify(listener);
    }