OperatorImpl { void registerNextOperator(OperatorImpl<RM, ?> nextOperator) { if (!initialized) { throw new IllegalStateException( String.format("Attempted to register next operator before initializing operator %s.", getOpImplId())); } this.registeredOperators.add(nextOperator); nextOperator.registerPrevOperator(this); }  final void init(InternalTaskContext internalTaskContext); final CompletionStage<Void> onMessageAsync(M message, MessageCollector collector,
      TaskCoordinator coordinator); final CompletionStage<Void> onTimer(MessageCollector collector, TaskCoordinator coordinator); final CompletionStage<Void> aggregateEndOfStream(EndOfStreamMessage eos, SystemStreamPartition ssp, MessageCollector collector,
      TaskCoordinator coordinator); final CompletionStage<Void> aggregateWatermark(WatermarkMessage watermarkMessage, SystemStreamPartition ssp,
      MessageCollector collector, TaskCoordinator coordinator); void close();  }
@Test(expected = IllegalStateException.class) public void testRegisterNextOperatorBeforeInitShouldThrow() { OperatorImpl<Object, Object> opImpl = new TestOpImpl(mock(Object.class)); opImpl.registerNextOperator(mock(OperatorImpl.class)); }