class TimestampedCacheFlushListener implements CacheFlushListener<K, ValueAndTimestamp<V>> {

    @Override
    public void apply(final K key,
                      final ValueAndTimestamp<V> newValue,
                      final ValueAndTimestamp<V> oldValue,
                      final long timestamp) {
        final ProcessorNode prev = context.currentNode();
        context.setCurrentNode(myNode);
        try {
            context.forward(
                key,
                new Change<>(getValueOrNull(newValue), getValueOrNull(oldValue)),
                To.all().withTimestamp(newValue != null ? newValue.timestamp() : timestamp));
        } finally {
            context.setCurrentNode(prev);
        }
    }

      TimestampedCacheFlushListener(final ProcessorContext context);

}