public void maybeForward(final Record<K, Change<V>> record) {
        if (!cachingEnabled) {
            if (sendOldValues) {
                context.forward(record);
            } else {
                context.forward(record.withValue(new Change<>(record.value().newValue, null, record.value().isLatest)));
            }
        }
    }
@Test
    public void shouldNotForwardRecordsIfWrappedStateStoreDoesCache() {
        @SuppressWarnings("unchecked")
        final WrappedStateStore<StateStore, String, String> store = mock(WrappedStateStore.class);
        @SuppressWarnings("unchecked")
        final InternalProcessorContext<String, Change<String>> context = mock(InternalProcessorContext.class);

        when(store.setFlushListener(null, false)).thenReturn(true);

        final TimestampedTupleForwarder<String, String> forwarder =
            new TimestampedTupleForwarder<>(
                store,
                (org.apache.kafka.streams.processor.api.ProcessorContext<String, Change<String>>) context,
                null,
                false
            );
        forwarder.maybeForward(new Record<>("key", new Change<>("newValue", "oldValue", true), 0L));
        forwarder.maybeForward(new Record<>("key", new Change<>("newValue", "oldValue", true), 42L));
    }