EndOfStreamStates { void update(EndOfStreamMessage eos, SystemStreamPartition ssp) { EndOfStreamState state = eosStates.get(ssp); state.update(eos.getTaskName()); } EndOfStreamStates(Set<SystemStreamPartition> ssps, Map<SystemStream, Integer> producerTaskCounts);   }
@Test public void testUpdate() { SystemStream input = new SystemStream("system", "input"); SystemStream intermediate = new SystemStream("system", "intermediate"); Set<SystemStreamPartition> ssps = new HashSet<>(); SystemStreamPartition inputPartition0 = new SystemStreamPartition(input, new Partition(0)); SystemStreamPartition intPartition0 = new SystemStreamPartition(intermediate, new Partition(0)); SystemStreamPartition intPartition1 = new SystemStreamPartition(intermediate, new Partition(1)); ssps.add(inputPartition0); ssps.add(intPartition0); ssps.add(intPartition1); Map<SystemStream, Integer> producerCounts = new HashMap<>(); producerCounts.put(intermediate, 2); EndOfStreamStates endOfStreamStates = new EndOfStreamStates(ssps, producerCounts); assertFalse(endOfStreamStates.isEndOfStream(input)); assertFalse(endOfStreamStates.isEndOfStream(intermediate)); assertFalse(endOfStreamStates.allEndOfStream()); IncomingMessageEnvelope envelope = IncomingMessageEnvelope.buildEndOfStreamEnvelope(inputPartition0); endOfStreamStates.update((EndOfStreamMessage) envelope.getMessage(), envelope.getSystemStreamPartition()); assertTrue(endOfStreamStates.isEndOfStream(input)); assertFalse(endOfStreamStates.isEndOfStream(intermediate)); assertFalse(endOfStreamStates.allEndOfStream()); EndOfStreamMessage eos = new EndOfStreamMessage("task 0"); endOfStreamStates.update(eos, intPartition0); endOfStreamStates.update(eos, intPartition1); assertFalse(endOfStreamStates.isEndOfStream(intermediate)); assertFalse(endOfStreamStates.allEndOfStream()); eos = new EndOfStreamMessage("task 1"); endOfStreamStates.update(eos, intPartition0); endOfStreamStates.update(eos, intPartition1); assertTrue(endOfStreamStates.isEndOfStream(intermediate)); assertTrue(endOfStreamStates.allEndOfStream()); }