public boolean putIfDifferentValues(final K key,
                                        final ValueAndTimestamp<V> newValue,
                                        final byte[] oldSerializedValue) {
        try {
            return maybeMeasureLatency(
                () -> {
                    final byte[] newSerializedValue = serdes.rawValue(newValue);
                    if (ValueAndTimestampSerializer.valuesAreSameAndTimeIsIncreasing(oldSerializedValue, newSerializedValue)) {
                        return false;
                    } else {
                        wrapped().put(keyBytes(key), newSerializedValue);
                        return true;
                    }
                },
                time,
                putSensor
            );
        } catch (final ProcessorStateException e) {
            final String message = String.format(e.getMessage(), key, newValue);
            throw new ProcessorStateException(message, e);
        }
    }
@Test
    public void shouldNotPutIfSameValuesAndGreaterTimestamp() {
        setUp();
        init();

        metered.put(KEY, VALUE_AND_TIMESTAMP);
        final ValueAndTimestampSerde<String> stringSerde = new ValueAndTimestampSerde<>(Serdes.String());
        final byte[] encodedOldValue = stringSerde.serializer().serialize("TOPIC", VALUE_AND_TIMESTAMP);

        final ValueAndTimestamp<String> newValueAndTimestamp = ValueAndTimestamp.make("value", 98L);
        assertFalse(metered.putIfDifferentValues(KEY, newValueAndTimestamp, encodedOldValue));
    }