+ map[partitionForId(id)].remove(id);
+ }
+
+ @Override
+ public void resetStorageForPartition(int partition)
+ {
+ log.info(
+ "reseting storage for partition {}: dropping {} entries",
+ partition,
+ map[partition].size());
+ map[partition].clear();
+ }
+
+ private int partitionForId(long id)
+ {
+ String key = Long.toString(id);
+ return TransferPartitioner.computeHashForKey(key, numPartitions);