class SessionCacheFlushListener implements CacheFlushListener<Windowed<K>, V> {

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

      SessionCacheFlushListener(final ProcessorContext context);

}