@Test
  public void testAllWithUpdate() {
    Counter allsCounter = mock(Counter.class);
    when(this.keyValueStoreMetrics.alls()).thenReturn(allsCounter);

    // fill in a range values for two different prefixes, but leave out index 1 so it can be added later
    for (int i = 0; i < 10; i++) {
      if (i != 1) {
        this.inMemoryKeyValueStore.put(key(i), value(i));
      }
    }
    KeyValueIterator<byte[], byte[]> all = this.inMemoryKeyValueStore.all();
    assertEquals(9, Iterators.size(all));

    this.inMemoryKeyValueStore.delete(key(2)); // delete an entry
    this.inMemoryKeyValueStore.put(key(3), value(OTHER_VALUE_PREFIX, 3)); // update an entry
    this.inMemoryKeyValueStore.put(key(1), value(1)); // add a new entry
    all = this.inMemoryKeyValueStore.all();
    for (int i = 0; i < 10; i++) {
      if (i != 2) { // index 2 was deleted
        if (i == 3) { // index 3 has an updated value
          assertEntryEquals(key(i), value(OTHER_VALUE_PREFIX, 3), all.next());
        } else {
          // all other entries (including index 1 have the "normal" entries)
          assertEntryEquals(key(i), value(i), all.next());
        }
      }
    }
    assertFalse(all.hasNext());

    verify(allsCounter, times(2)).inc();
    // key size increments: 9 calls for iterator size check of first "all", 9 calls for second "all"
    verify(this.bytesReadCounter, times(18)).inc(DEFAULT_KEY_LENGTH);
    /*
     * value size increments: 9 calls for iterator size check of first "all", 8 calls for second "all" (updated entry is
     * different)
     */
    verify(this.bytesReadCounter, times(17)).inc(DEFAULT_VALUE_LENGTH);
    // 1 call for "updatedValue"
    verify(this.bytesReadCounter).inc(value(OTHER_VALUE_PREFIX, 3).length);
  }
The length of the result of {@link #value(int)} will always be the same, so this can be used as the length for any
key produced by {@link #value(int)}.
samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/TestInMemoryKeyValueStore.java