+
+ Integer partition = record.partition();
+ String key = record.key() == null ? "NULL" : record.key();
+
+ if (!seen.containsKey(partition))
+ seen.put(partition, new HashMap<>());
+
+ Map<String, Integer> byKey = seen.get(partition);
+
+ if (!byKey.containsKey(key))
+ byKey.put(key, 0);
+
+ int seenByKey = byKey.get(key);
+ seenByKey++;
+ byKey.put(key, seenByKey);