@Override
    public Entry<byte[], byte[]> next() {
      Map.Entry<byte[], byte[]> n = this.iter.next();
      if (n != null && n.getKey() != null) {
        this.metrics.bytesRead().inc(n.getKey().length);
      }
      if (n != null && n.getValue() != null) {
        this.metrics.bytesRead().inc(n.getValue().length);
      }
      return new Entry<>(n.getKey(), n.getValue());
    }
@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