import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
private boolean isChainable(StreamEdge edge, boolean isChainingEnabled) {
	StreamNode upStreamVertex = edge.getSourceVertex();
	StreamNode downStreamVertex = edge.getTargetVertex();

	StreamOperator<?> headOperator = upStreamVertex.getOperator();
	StreamOperator<?> outOperator = downStreamVertex.getOperator();

	return downStreamVertex.getInEdges().size() == 1
			&& outOperator != null
			&& headOperator != null
			&& upStreamVertex.isSameSlotSharingGroup(downStreamVertex)
			&& outOperator.getChainingStrategy() == ChainingStrategy.ALWAYS
			&& (headOperator.getChainingStrategy() == ChainingStrategy.HEAD ||
			headOperator.getChainingStrategy() == ChainingStrategy.ALWAYS)
			&& (edge.getPartitioner() instanceof ForwardPartitioner)
			&& upStreamVertex.getParallelism() == downStreamVertex.getParallelism()
			&& isChainingEnabled;
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
public static boolean isChainable(StreamEdge edge, StreamGraph streamGraph) {
	StreamNode upStreamVertex = edge.getSourceVertex();
	StreamNode downStreamVertex = edge.getTargetVertex();

	StreamOperator<?> headOperator = upStreamVertex.getOperator();
	StreamOperator<?> outOperator = downStreamVertex.getOperator();

	return downStreamVertex.getInEdges().size() == 1
			&& outOperator != null
			&& headOperator != null
			&& upStreamVertex.isSameSlotSharingGroup(downStreamVertex)
			&& outOperator.getChainingStrategy() == ChainingStrategy.ALWAYS
			&& (headOperator.getChainingStrategy() == ChainingStrategy.HEAD ||
				headOperator.getChainingStrategy() == ChainingStrategy.ALWAYS)
			&& (edge.getPartitioner() instanceof ForwardPartitioner)
			&& upStreamVertex.getParallelism() == downStreamVertex.getParallelism()
			&& streamGraph.isChainingEnabled();
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
public DoFnOperator(
    DoFn<InputT, OutputT> doFn,
    String stepName,
    Coder<WindowedValue<InputT>> inputCoder,
    TupleTag<OutputT> mainOutputTag,
    List<TupleTag<?>> additionalOutputTags,
    OutputManagerFactory<OutputT> outputManagerFactory,
    WindowingStrategy<?, ?> windowingStrategy,
    Map<Integer, PCollectionView<?>> sideInputTagMapping,
    Collection<PCollectionView<?>> sideInputs,
    PipelineOptions options,
    Coder<?> keyCoder) {
  this.doFn = doFn;
  this.stepName = stepName;
  this.inputCoder = inputCoder;
  this.mainOutputTag = mainOutputTag;
  this.additionalOutputTags = additionalOutputTags;
  this.sideInputTagMapping = sideInputTagMapping;
  this.sideInputs = sideInputs;
  this.serializedOptions = new SerializablePipelineOptions(options);
  this.windowingStrategy = windowingStrategy;
  this.outputManagerFactory = outputManagerFactory;

  setChainingStrategy(ChainingStrategy.ALWAYS);

  this.keyCoder = keyCoder;

  this.timerCoder =
      TimerInternals.TimerDataCoder.of(windowingStrategy.getWindowFn().windowCoder());

  FlinkPipelineOptions flinkOptions = options.as(FlinkPipelineOptions.class);

  this.maxBundleSize = flinkOptions.getMaxBundleSize();
  this.maxBundleTimeMills = flinkOptions.getMaxBundleTimeMills();
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
/**
 * Note: this test fails if we don't check for exceptions in the source contexts and do not
 * synchronize in the source contexts.
 */
@Test
public void testOperatorChainedToSource() throws Exception {

	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	env.setStreamTimeCharacteristic(timeCharacteristic);
	env.setParallelism(1);

	DataStream<String> source = env.addSource(new InfiniteTestSource());

	source.transform("Custom Operator", BasicTypeInfo.STRING_TYPE_INFO, new TimerOperator(ChainingStrategy.ALWAYS));

	boolean testSuccess = false;
	try {
		env.execute("Timer test");
	} catch (JobExecutionException e) {
		if (e.getCause() instanceof TimerException) {
			TimerException te = (TimerException) e.getCause();
			if (te.getCause() instanceof RuntimeException) {
				RuntimeException re = (RuntimeException) te.getCause();
				if (re.getMessage().equals("TEST SUCCESS")) {
					testSuccess = true;
				} else {
					throw e;
				}
			} else {
				throw e;
			}
		} else {
			throw e;
		}
	}
	Assert.assertTrue(testSuccess);
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
/**
 * Note: this test fails if we don't check for exceptions in the source contexts and do not
 * synchronize in the source contexts.
 */
@Test
public void testOneInputOperatorWithoutChaining() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	env.setStreamTimeCharacteristic(timeCharacteristic);
	env.setParallelism(1);

	DataStream<String> source = env.addSource(new InfiniteTestSource());

	source.transform("Custom Operator", BasicTypeInfo.STRING_TYPE_INFO, new TimerOperator(ChainingStrategy.NEVER));

	boolean testSuccess = false;
	try {
		env.execute("Timer test");
	} catch (JobExecutionException e) {
		if (e.getCause() instanceof TimerException) {
			TimerException te = (TimerException) e.getCause();
			if (te.getCause() instanceof RuntimeException) {
				RuntimeException re = (RuntimeException) te.getCause();
				if (re.getMessage().equals("TEST SUCCESS")) {
					testSuccess = true;
				} else {
					throw e;
				}
			} else {
				throw e;
			}
		} else {
			throw e;
		}
	}
	Assert.assertTrue(testSuccess);
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
@Test
public void testTwoInputOperatorWithoutChaining() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	env.setStreamTimeCharacteristic(timeCharacteristic);
	env.setParallelism(1);

	DataStream<String> source = env.addSource(new InfiniteTestSource());

	source.connect(source).transform(
			"Custom Operator",
			BasicTypeInfo.STRING_TYPE_INFO,
			new TwoInputTimerOperator(ChainingStrategy.NEVER));

	boolean testSuccess = false;
	try {
		env.execute("Timer test");
	} catch (JobExecutionException e) {
		if (e.getCause() instanceof TimerException) {
			TimerException te = (TimerException) e.getCause();
			if (te.getCause() instanceof RuntimeException) {
				RuntimeException re = (RuntimeException) te.getCause();
				if (re.getMessage().equals("TEST SUCCESS")) {
					testSuccess = true;
				} else {
					throw e;
				}
			} else {
				throw e;
			}
		} else {
			throw e;
		}
	}
	Assert.assertTrue(testSuccess);
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
public AsyncWaitOperator(
		AsyncFunction<IN, OUT> asyncFunction,
		long timeout,
		int capacity,
		AsyncDataStream.OutputMode outputMode) {
	super(asyncFunction);
	chainingStrategy = ChainingStrategy.ALWAYS;

	Preconditions.checkArgument(capacity > 0, "The number of concurrent async operation should be greater than 0.");
	this.capacity = capacity;

	this.outputMode = Preconditions.checkNotNull(outputMode, "outputMode");

	this.timeout = timeout;
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
/**
 * Creates a new {@code WindowOperator} based on the given policies and user functions.
 */
public WindowOperator(
		WindowAssigner<? super IN, W> windowAssigner,
		TypeSerializer<W> windowSerializer,
		KeySelector<IN, K> keySelector,
		TypeSerializer<K> keySerializer,
		StateDescriptor<? extends AppendingState<IN, ACC>, ?> windowStateDescriptor,
		InternalWindowFunction<ACC, OUT, K, W> windowFunction,
		Trigger<? super IN, ? super W> trigger,
		long allowedLateness,
		OutputTag<IN> lateDataOutputTag) {

	super(windowFunction);

	checkArgument(!(windowAssigner instanceof BaseAlignedWindowAssigner),
		"The " + windowAssigner.getClass().getSimpleName() + " cannot be used with a WindowOperator. " +
			"This assigner is only used with the AccumulatingProcessingTimeWindowOperator and " +
			"the AggregatingProcessingTimeWindowOperator");

	checkArgument(allowedLateness >= 0);

	checkArgument(windowStateDescriptor == null || windowStateDescriptor.isSerializerInitialized(),
			"window state serializer is not properly initialized");

	this.windowAssigner = checkNotNull(windowAssigner);
	this.windowSerializer = checkNotNull(windowSerializer);
	this.keySelector = checkNotNull(keySelector);
	this.keySerializer = checkNotNull(keySerializer);
	this.windowStateDescriptor = windowStateDescriptor;
	this.trigger = checkNotNull(trigger);
	this.allowedLateness = allowedLateness;
	this.lateDataOutputTag = lateDataOutputTag;

	setChainingStrategy(ChainingStrategy.ALWAYS);
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
/**
 * Creates an event time-based reordering operator.
 */
public EventTimeOrderingOperator() {
    chainingStrategy = ChainingStrategy.ALWAYS;
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
@Override
public DataStreamSink<T> disableChaining() {
	this.transformation.setChainingStrategy(ChainingStrategy.NEVER);
	return this;
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
public CustomOperator(boolean timestampsEnabled) {
	setChainingStrategy(ChainingStrategy.ALWAYS);
	this.timestampsEnabled = timestampsEnabled;
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
public TimestampCheckingOperator() {
	setChainingStrategy(ChainingStrategy.ALWAYS);
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
public TimerOperator(ChainingStrategy chainingStrategy) {
	setChainingStrategy(chainingStrategy);
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
public TwoInputTimerOperator(ChainingStrategy chainingStrategy) {
	setChainingStrategy(chainingStrategy);
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
@Override
public final void setChainingStrategy(ChainingStrategy strategy) {
	throw new UnsupportedOperationException("Cannot set chaining strategy on Split Transformation.");
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
@Override
public final void setChainingStrategy(ChainingStrategy strategy) {
	operator.setChainingStrategy(strategy);
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
@Override
public final void setChainingStrategy(ChainingStrategy strategy) {
	throw new UnsupportedOperationException("Cannot set chaining strategy on Union Transformation.");
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
@Override
public final void setChainingStrategy(ChainingStrategy strategy) {
	throw new UnsupportedOperationException("Cannot set chaining strategy on SideOutput Transformation.");
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
@Override
public final void setChainingStrategy(ChainingStrategy strategy) {
	throw new UnsupportedOperationException("Cannot set chaining strategy on Select Transformation.");
}
 

import org.apache.flink.streaming.api.operators.ChainingStrategy; //导入依赖的package包/类
public TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner) {
	super(assigner);
	this.chainingStrategy = ChainingStrategy.ALWAYS;
}