[FLINK-12297] Make ClosureCleaner recursive

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.8.0
• Fix Version/s: 1.8.1, 1.9.0
• Component/s: API / DataSet, API / DataStream
• Labels: pull-request-available
• Assignee: WenJun Min
• Reporter: Dawid Wysakowicz
• Created: 23/Apr/19 06:54
• Updated: 01/Oct/19 15:39
• Resolved: 15/Jun/19 06:57

Description

Right now we do not invoke closure cleaner on output tags. Therefore such code:

	@Test
	public void testFlatSelectSerialization() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Integer> elements = env.fromElements(1, 2, 3);
		OutputTag<Integer> outputTag = new OutputTag<Integer>("AAA") {};
		CEP.pattern(elements, Pattern.begin("A")).flatSelect(
			outputTag,
			new PatternFlatTimeoutFunction<Integer, Integer>() {
				@Override
				public void timeout(
					Map<String, List<Integer>> pattern,
					long timeoutTimestamp,
					Collector<Integer> out) throws Exception {

				}
			},
			new PatternFlatSelectFunction<Integer, Object>() {
				@Override
				public void flatSelect(Map<String, List<Integer>> pattern, Collector<Object> out) throws Exception {

				}
			}
		);

		env.execute();
	}

will fail with {{The implementation of the PatternFlatSelectAdapter is not serializable. }} exception

Issue Links

• causes: FLINK-13586 Method ClosureCleaner.clean broke backward compatibility between 1.8.0 and 1.8.1
• links to: GitHub Pull Request #8280
• links to: GitHub Pull Request #8744
• links to: GitHub Pull Request #9069