SizeByPartitionFunction { public Map<TopicPartition, Long> apply(Map<TopicPartition, Replica> partitionsAndLogDir) { Map<TopicPartition, Long> result = new HashMap<>(); for (Map.Entry<TopicPartition, Replica> entry : partitionsAndLogDir.entrySet()) { TopicPartition topicPartition = entry.getKey(); Replica replica = entry.getValue(); Path path = new File(replica.getLogDir(), topicPartition.toString()).toPath(); try { long size = Files.walk(path).filter(Files::isRegularFile).mapToLong(p -> p.toFile().length()).sum(); result.put(topicPartition, size); } catch (IOException e) { throw new UncheckedIOException("Unable to determine size of " + path, e); } } return result; }  Map<TopicPartition, Long> apply(Map<TopicPartition, Replica> partitionsAndLogDir);  }
@Test public void typical() throws Exception { TopicPartition topicPartition = new TopicPartition("topic", 0); File folder = temp.newFolder(topicPartition.toString()); wrap(new byte[] { 0 }).copyTo(asByteSink(new File(folder, "data"))); Map<TopicPartition, Replica> partitionsAndLogDir = singletonMap(topicPartition, new Replica(temp.getRoot().getAbsolutePath(), 0L)); Map<TopicPartition, Long> result = underTest.apply(partitionsAndLogDir); assertThat(result.size(), is(1)); long size = result.get(topicPartition); assertThat(size, is(1L)); }
@Test(expected = UncheckedIOException.class) public void nonExistentPath() throws Exception { TopicPartition topicPartition = new TopicPartition("topic", 0); Map<TopicPartition, Replica> partitionsAndLogDir = singletonMap(topicPartition, new Replica("non-existent-folder", 0L)); underTest.apply(partitionsAndLogDir); }