Problem

When JobManager starts, the ExecutionJobVertex constructor calls the AbstractKuduInputFormat#createInputSplits method, Causes the KuduReader object to create the KuduClient and KuduSession objects, but ExecutionJobVertex does not call the Kudureader #close() method with the display, causing the Kudu-related handle to remain open after the Flink task is closed. As a result, the number of server handles increases with the number of Flink tasks. As a result, the entire server becomes unavailable

Code

ExecutionJobVertex#constructor

AbstractKuduInputFormat#createInputSplits{}

AbstractKuduInputFormat#startKuduReader

{}