Key mapping about Stateless Operator

[maxCodeVector]

Since we assume that each operator has its own key, so currently, we add "keyby" after each operator except for the sink.

It is a problem when rescaling such an operator. Because in our design, rescale a stateless operator does not need synchronization. However, that will cause the operator to receive a key that does not belong to it.

In flink, when processing data, it will extract key using the code below:

		@Override
		public void emitRecord(StreamRecord<IN> record) throws Exception {
			synchronized (lock) {
				numRecordsIn.inc();
				operator.setKeyContextElement1(record);
				operator.processElement(record);
			}
		}

When receiving other's keys, an exception will be thrown.