一文搞懂 flink state key 的设置方式

@Override
	protected void run() throws Exception {
		// cache processor reference on the stack, to make the code more JIT friendly
		final StreamInputProcessor<IN> inputProcessor = this.inputProcessor;
		//处理输入的消息
		while (running && inputProcessor.processInput()) {
			// all the work happens in the "processInput" method
		}
	}

						// now we can do the actual processing
						StreamRecord<IN> record = recordOrMark.asRecord();
						synchronized (lock) {
							numRecordsIn.inc();
							//set KeyContext setCurrentKey
			//设置 keyContext (提供了用来查询和设置 keyed operation 的 current key 的接口)
							streamOperator.setKeyContextElement1(record);
			//这里开始调用用户自己的代码
							streamOperator.processElement(record);
						}

//自定义的 KeySelector 在此处起作用
	private <T> void setKeyContextElement(StreamRecord<T> record, KeySelector<T, ?> selector) throws Exception {
		if (selector != null) {
			Object key = selector.getKey(record.getValue());
			setCurrentKey(key);
		}
	}

	@SuppressWarnings({"unchecked", "rawtypes"})
	public void setCurrentKey(Object key) {
		if (keyedStateBackend != null) {
			try {
				// need to work around type restrictions
				@SuppressWarnings("unchecked,rawtypes")
				AbstractKeyedStateBackend rawBackend = (AbstractKeyedStateBackend) keyedStateBackend;
				
				//设置 keyedStateBackend currentKey
				rawBackend.setCurrentKey(key);
			} catch (Exception e) {
				throw new RuntimeException("Exception occurred while setting the current key context.", e);
			}
		}
	}

@Override
	public void setCurrentKey(K newKey) {
		super.setCurrentKey(newKey);
		// 每个 key 都会调用一次 将 key group and key 写入 byte[] 中，每次开始写入前都会清空，后续 state 的操作都会从这个 byte[] 中读
		sharedRocksKeyBuilder.setKeyAndKeyGroup(getCurrentKey(), getCurrentKeyGroupIndex());
	}

@Nonnull
	public <N> byte[] buildCompositeKeyNamespace(@Nonnull N namespace, @Nonnull TypeSerializer<N> namespaceSerializer) {
		try {
			// 每次真正操作时候，serializeNamespace
			serializeNamespace(namespace, namespaceSerializer);
			// 将已序列化的 key_group,key,namespace 作为一个整体 copy 出来,这也就是 state key
			final byte[] result = keyOutView.getCopyOfBuffer();
			// 重置，类似于重置游标，去除 namespace bytes
			resetToKey();
			return result;
		} catch (IOException shouldNeverHappen) {
			throw new FlinkRuntimeException(shouldNeverHappen);
		}
	}