@Override
  public void put(byte[] key, byte[] value) {
    this.metrics.puts().inc();
    Preconditions.checkArgument(key != null, "Null argument 'key' not allowed");
    if (value == null) {
      this.metrics.deletes().inc();
      this.underlying.remove(key);
    } else {
      this.metrics.bytesWritten().inc(key.length + value.length);
      this.underlying.put(key, value);
    }
  }
@param metrics A metrics instance to publish key-value store related statistics
samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryKeyValueStore.java