OperatorImpl { public final void init(InternalTaskContext internalTaskContext) { final Context context = internalTaskContext.getContext(); String opId = getOpImplId(); if (initialized) { throw new IllegalStateException(String.format("Attempted to initialize Operator %s more than once.", opId)); } if (closed) { throw new IllegalStateException(String.format("Attempted to initialize Operator %s after it was closed.", opId)); } this.highResClock = createHighResClock(context.getJobContext().getConfig()); registeredOperators = new HashSet<>(); prevOperators = new HashSet<>(); inputStreams = new HashSet<>(); final ContainerContext containerContext = context.getContainerContext(); final MetricsRegistry metricsRegistry = containerContext.getContainerMetricsRegistry(); this.numMessage = metricsRegistry.newCounter(METRICS_GROUP, opId + "-messages"); this.handleMessageNs = metricsRegistry.newTimer(METRICS_GROUP, opId + "-handle-message-ns"); this.handleTimerNs = metricsRegistry.newTimer(METRICS_GROUP, opId + "-handle-timer-ns"); final TaskContext taskContext = context.getTaskContext(); this.taskName = taskContext.getTaskModel().getTaskName(); this.eosStates = (EndOfStreamStates) internalTaskContext.fetchObject(EndOfStreamStates.class.getName()); this.watermarkStates = (WatermarkStates) internalTaskContext.fetchObject(WatermarkStates.class.getName()); this.controlMessageSender = new ControlMessageSender(internalTaskContext.getStreamMetadataCache()); this.taskModel = taskContext.getTaskModel(); this.callbackScheduler = taskContext.getCallbackScheduler(); handleInit(context); initialized = true; }  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 testMultipleInitShouldThrow() { OperatorImpl<Object, Object> opImpl = new TestOpImpl(mock(Object.class)); opImpl.init(this.internalTaskContext); opImpl.init(this.internalTaskContext); }