docs: @Override
  	public void setup(StreamTask<?, ?> containingTask, StreamConfig config, Output<StreamRecord<OUT>> output) {
  		final Environment environment = containingTask.getEnvironment();
  		this.container = containingTask;
  		this.config = config;
  		try {
  			OperatorMetricGroup operatorMetricGroup = environment.getMetricGroup().getOrAddOperator(config.getOperatorID(), config.getOperatorName());
  			this.output = new CountingOutput(output, operatorMetricGroup.getIOMetricGroup().getNumRecordsOutCounter());
  			if (config.isChainStart()) {
  				operatorMetricGroup.getIOMetricGroup().reuseInputMetricsForTask();
  			}
  			if (config.isChainEnd()) {
  				operatorMetricGroup.getIOMetricGroup().reuseOutputMetricsForTask();
  			}
  			this.metrics = operatorMetricGroup;
  		} catch (Exception e) {
  			LOG.warn("An error occurred while instantiating task metrics.", e);
  			this.metrics = UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup();
  			this.output = output;
  		}
  
  		try {
  			Configuration taskManagerConfig = environment.getTaskManagerInfo().getConfiguration();
  			int historySize = taskManagerConfig.getInteger(MetricOptions.LATENCY_HISTORY_SIZE);
  			if (historySize <= 0) {
  				LOG.warn("{} has been set to a value equal or below 0: {}. Using default.", MetricOptions.LATENCY_HISTORY_SIZE, historySize);
  				historySize = MetricOptions.LATENCY_HISTORY_SIZE.defaultValue();
  			}
  
  			final String configuredGranularity = taskManagerConfig.getString(MetricOptions.LATENCY_SOURCE_GRANULARITY);
  			LatencyStats.Granularity granularity;
  			try {
  				granularity = LatencyStats.Granularity.valueOf(configuredGranularity.toUpperCase(Locale.ROOT));
  			} catch (IllegalArgumentException iae) {
  				granularity = LatencyStats.Granularity.OPERATOR;
  				LOG.warn(
  					"Configured value {} option for {} is invalid. Defaulting to {}.",
  					configuredGranularity,
  					MetricOptions.LATENCY_SOURCE_GRANULARITY.key(),
  					granularity);
  			}
  			TaskManagerJobMetricGroup jobMetricGroup = this.metrics.parent().parent();
  			this.latencyStats = new LatencyStats(jobMetricGroup.addGroup("latency"),
  				historySize,
  				container.getIndexInSubtaskGroup(),
  				getOperatorID(),
  				granularity);
  		} catch (Exception e) {
  			LOG.warn("An error occurred while instantiating latency metrics.", e);
  			this.latencyStats = new LatencyStats(
  				UnregisteredMetricGroups.createUnregisteredTaskManagerJobMetricGroup().addGroup("latency"),
  				1,
  				0,
  				new OperatorID(),
  				LatencyStats.Granularity.SINGLE);
  		}
  
  		this.runtimeContext = new StreamingRuntimeContext(this, environment, container.getAccumulatorMap());
  
  		stateKeySelector1 = config.getStatePartitioner(0, getUserCodeClassloader());
  		stateKeySelector2 = config.getStatePartitioner(1, getUserCodeClassloader());
  	}