StartpointManager { public void deleteAllStartpoints() { Set<String> readWriteKeys = readWriteStore.all().keySet(); for (String key : readWriteKeys) { readWriteStore.delete(key); } readWriteStore.flush(); } StartpointManager(MetadataStore metadataStore); void start(); void stop(); void writeStartpoint(SystemStreamPartition ssp, Startpoint startpoint); void writeStartpoint(SystemStreamPartition ssp, TaskName taskName, Startpoint startpoint); @VisibleForTesting Optional<Startpoint> readStartpoint(SystemStreamPartition ssp); @VisibleForTesting Optional<Startpoint> readStartpoint(SystemStreamPartition ssp, TaskName taskName); Optional<Startpoint> readStartpoint(Map<String, byte[]> startpointMap, SystemStreamPartition ssp, TaskName taskName); void deleteStartpoint(SystemStreamPartition ssp); void deleteStartpoint(SystemStreamPartition ssp, TaskName taskName); void deleteAllStartpoints(); Map<TaskName, Map<SystemStreamPartition, Startpoint>> fanOut(Map<TaskName, Set<SystemStreamPartition>> taskToSSPs); Map<SystemStreamPartition, Startpoint> getFanOutForTask(TaskName taskName); void removeFanOutForTask(TaskName taskName); void removeAllFanOuts(); static final String NAMESPACE; }
@Test public void testDeleteAllStartpoints() throws IOException { SystemStreamPartition sspBroadcast = new SystemStreamPartition("mockSystem1", "mockStream1", new Partition(2)); SystemStreamPartition sspSingle = new SystemStreamPartition("mockSystem2", "mockStream2", new Partition(3)); TaskName taskWithNonBroadcast = new TaskName("t1"); List<TaskName> tasks = ImmutableList.of(new TaskName("t0"), taskWithNonBroadcast, new TaskName("t2"), new TaskName("t3"), new TaskName("t4"), new TaskName("t5")); Map<TaskName, Set<SystemStreamPartition>> taskToSSPs = tasks.stream() .collect(Collectors .toMap(task -> task, task -> task.equals(taskWithNonBroadcast) ? ImmutableSet.of(sspBroadcast, sspSingle) : ImmutableSet.of(sspBroadcast))); StartpointSpecific startpoint42 = new StartpointSpecific("42"); startpointManager.writeStartpoint(sspBroadcast, startpoint42); startpointManager.writeStartpoint(sspSingle, startpoint42); Map<TaskName, Map<SystemStreamPartition, Startpoint>> tasksFannedOutTo = startpointManager.fanOut(taskToSSPs); Assert.assertEquals(tasks.size(), tasksFannedOutTo.size()); Assert.assertTrue(tasksFannedOutTo.keySet().containsAll(tasks)); Assert.assertFalse("Should be deleted after fan out", startpointManager.readStartpoint(sspBroadcast).isPresent()); Assert.assertFalse("Should be deleted after fan out", startpointManager.readStartpoint(sspSingle).isPresent()); startpointManager.writeStartpoint(sspBroadcast, startpoint42); startpointManager.writeStartpoint(sspSingle, startpoint42); Assert.assertEquals(2, startpointManager.getReadWriteStore().all().size()); startpointManager.deleteAllStartpoints(); Assert.assertEquals(0, startpointManager.getReadWriteStore().all().size()); Assert.assertEquals(tasks.size(), startpointManager.getFanOutStore().all().size()); }