public final void init(InternalTaskContext internalTaskContext) {
    final Context context = internalTaskContext.getContext();
    final Config config = context.getJobContext().getConfig();

    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(config);
    registeredOperators = new LinkedHashSet<>();
    prevOperators = new LinkedHashSet<>();
    inputStreams = new LinkedHashSet<>();

    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.drainStates = (DrainStates) internalTaskContext.fetchObject(DrainStates.class.getName());

    this.controlMessageSender = new ControlMessageSender(internalTaskContext.getStreamMetadataCache());
    this.taskModel = taskContext.getTaskModel();
    this.callbackScheduler = taskContext.getCallbackScheduler();
    handleInit(context);
    JobConfig jobConfig = new JobConfig(config);
    this.elasticityFactor = jobConfig.getElasticityFactor();
    this.operatorExecutorEnabled = jobConfig.getOperatorFrameworkExecutorEnabled();
    this.operatorExecutor = context.getTaskContext().getOperatorExecutor();

    initialized = true;
  }
Initialize this {@link OperatorImpl} and its user-defined functions.

@param internalTaskContext the {@link InternalTaskContext} for the task
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java