flink1.15.3 jvm metaspace一直上涨不会释放直至溢出

["junjie.miao@goupwith.com" <ju...@goupwith.com>]

目前我们使用flink local模式用于开发平台的flink sql预览调试，发现一个问题是JVM metaspace一直上涨不会被回收。
正常情况应该是每次执行完，其FlinkUserCodeClassLoader中所有加载的类会被gc所回收，通过本地代码调试发现flink自身这块应该有BUG。

flink版本为1.15.3

测试代码如下，metaspace会一直上涨永远不会被gc回收。
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.ArrayList;
import java.util.List;

public class FlinkSqlTest {

    private static String createSql = "CREATE TEMPORARY TABLE datagen (\n" +
            "  `f1` STRING,\n" +
            "  `f2` INTEGER,\n" +
            "  `f3` TIMESTAMP(3),\n" +
            "  `pt` AS PROCTIME()\n" +
            ") WITH (\n" +
            "  'connector' = 'datagen',\n" +
            "  'rows-per-second' = '10'\n" +
            ", 'number-of-rows' = '100'" +
            ")";

    private static String selectSql = "select * from datagen";

    public static void main(String[] args) {
        for (int i = 0; i < 50; i++) {
            Configuration configuration = new Configuration();
            configuration.set(CoreOptions.DEFAULT_PARALLELISM, 1);
//            configuration.set(CoreOptions.CLASSLOADER_RESOLVE_ORDER, "parent-first");
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
            StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
            List<TableResult> tableResults = new ArrayList<>();
            tableResults.add(tEnv.executeSql(createSql));
            tableResults.add(tEnv.executeSql(selectSql));
            tableResults.forEach(tableResult -> tableResult.getJobClient().ifPresent(jobClient -> {
                try {
                    jobClient.getJobExecutionResult().get();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }));
            System.out.println("-----------------" + i + "执行结束----------------");
        }
        try {
            Thread.sleep(24 * 60 * 60 * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("-----------------休眠结束----------------");
    }

}