@Override
    public void apply(final Record<KOut, Change<ValueAndTimestamp<VOut>>> record) {
        @SuppressWarnings("rawtypes") final ProcessorNode prev = context.currentNode();
        context.setCurrentNode(myNode);
        try {
            context.forward(
                record
                    .withValue(
                        new Change<>(
                            getValueOrNull(record.value().newValue),
                            getValueOrNull(record.value().oldValue),
                            record.value().isLatest))
                    .withTimestamp(
                        record.value().newValue != null ? record.value().newValue.timestamp()
                            : record.timestamp())
            );
        } finally {
            context.setCurrentNode(prev);
        }
    }
@Test
    public void shouldForwardValueTimestampIfNewValueExists() {
        @SuppressWarnings("unchecked")
        final InternalProcessorContext<String, Change<String>> context = mock(InternalProcessorContext.class);
        doNothing().when(context).forward(
            new Record<>(
                "key",
                new Change<>("newValue", "oldValue"),
                42L));

        new TimestampedCacheFlushListener<>(context).apply(
            new Record<>(
                "key",
                new Change<>(
                    ValueAndTimestamp.make("newValue", 42L),
                    ValueAndTimestamp.make("oldValue", 21L)),
                73L));

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