[FLINK-30906] TwoInputStreamTask passes wrong configuration object when creating input processor

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.17.0, 1.16.1
• Fix Version/s: 1.19.0
• Component/s: Runtime / Task
• Labels: pull-request-available
• Assignee: Yun Gao
• Reporter: Yun Gao
• Created: 06/Feb/23 07:21
• Updated: 13/Sep/23 10:26
• Resolved: 13/Sep/23 10:26

Description

It seems StreamTwoInputProcessorFactory.create is passed with wrong configuration object: the taskManagerConfiguration should be __ 
getEnvironment().getTaskManagerInfo().getConfiguration(). 
 
And in the following logic, it seems to indeed try to load taskmanager options from this configuration object, like state-backend and taskmanager.memory.managed.consumer-weights 
 

[1]https://github.com/apache/flink/blob/111342f37bdc0d582d3f7af458d9869f0548299f/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java#L98

Issue Links

• links to: GitHub Pull Request #23401