@Override
    public void put(final K key,
                    final V value) {
        Objects.requireNonNull(key, "key cannot be null");
        try {
            maybeMeasureLatency(() -> wrapped().put(keyBytes(key), serdes.rawValue(value)), time, putSensor);
            maybeRecordE2ELatency();
        } catch (final ProcessorStateException e) {
            final String message = String.format(e.getMessage(), key, value);
            throw new ProcessorStateException(message, e);
        }
    }
@Test
    public void shouldThrowNullPointerOnPutIfKeyIsNull() {
        setUpWithoutContext();
        assertThrows(NullPointerException.class, () -> metered.put(null, VALUE));
    }