OperatorImplGraph { public void close() { List<OperatorImpl> initializationOrder = new ArrayList<>(operatorImpls.values()); List<OperatorImpl> finalizationOrder = Lists.reverse(initializationOrder); finalizationOrder.forEach(OperatorImpl::close); } OperatorImplGraph(OperatorSpecGraph specGraph, Context context, Clock clock); InputOperatorImpl getInputOperator(SystemStream systemStream); void close(); Collection<InputOperatorImpl> getAllInputOperators();  }
@Test public void testOperatorGraphInitAndClose() { String inputStreamId1 = "input1"; String inputStreamId2 = "input2"; String inputSystem = "input-system"; TaskName mockTaskName = mock(TaskName.class); TaskModel taskModel = mock(TaskModel.class); when(taskModel.getTaskName()).thenReturn(mockTaskName); when(this.context.getTaskContext().getTaskModel()).thenReturn(taskModel); StreamApplicationDescriptorImpl graphSpec = new StreamApplicationDescriptorImpl(appDesc -> { GenericSystemDescriptor sd = new GenericSystemDescriptor(inputSystem, "mockFactoryClass"); GenericInputDescriptor inputDescriptor1 = sd.getInputDescriptor(inputStreamId1, mock(Serde.class)); GenericInputDescriptor inputDescriptor2 = sd.getInputDescriptor(inputStreamId2, mock(Serde.class)); MessageStream<Object> inputStream1 = appDesc.getInputStream(inputDescriptor1); MessageStream<Object> inputStream2 = appDesc.getInputStream(inputDescriptor2); Function mapFn = (Function & Serializable) m -> m; inputStream1.map(new TestMapFunction<Object, Object>("1", mapFn)) .map(new TestMapFunction<Object, Object>("2", mapFn)); inputStream2.map(new TestMapFunction<Object, Object>("3", mapFn)) .map(new TestMapFunction<Object, Object>("4", mapFn)); }, getConfig()); OperatorImplGraph opImplGraph = new OperatorImplGraph(graphSpec.getOperatorSpecGraph(), this.context, SystemClock.instance()); List<String> initializedOperators = BaseTestFunction.getInitListByTaskName(mockTaskName); assertEquals(initializedOperators.get(0), "1"); assertEquals(initializedOperators.get(1), "2"); assertEquals(initializedOperators.get(2), "3"); assertEquals(initializedOperators.get(3), "4"); opImplGraph.close(); List<String> closedOperators = BaseTestFunction.getCloseListByTaskName(mockTaskName); assertEquals(closedOperators.get(0), "4"); assertEquals(closedOperators.get(1), "3"); assertEquals(closedOperators.get(2), "2"); assertEquals(closedOperators.get(3), "1"); }