@Override
    public void apply(final Record<Windowed<KOut>, Change<VOut>> record) {
        @SuppressWarnings("rawtypes") final ProcessorNode prev = context.currentNode();
        context.setCurrentNode(myNode);
        try {
            context.forward(record.withTimestamp(record.key().window().end()));
        } finally {
            context.setCurrentNode(prev);
        }
    }
@Test
    public void shouldForwardKeyNewValueOldValueAndTimestamp() {
        @SuppressWarnings("unchecked")
        final InternalProcessorContext<Windowed<String>, Change<String>> context = mock(InternalProcessorContext.class);
        doNothing().when(context).forward(
            new Record<>(
                new Windowed<>("key", new SessionWindow(21L, 73L)),
                new Change<>("newValue", "oldValue"),
                73L));

        new SessionCacheFlushListener<>(context).apply(
            new Record<>(
                new Windowed<>("key", new SessionWindow(21L, 73L)),
                new Change<>("newValue", "oldValue"),
                42L));

        verify(context, times(2)).setCurrentNode(null);
    }