title: java.lang.OutOfMemoryError: Metaspace in apache flink 1.10 deploy kubernetes cluster
body: <p>I am using apache flink 1.10 to consume RabbitMQ message and sink to MySQL using restful api, now the taskmanager exit and throw error:</p>
  <pre><code>2020-07-02 05:36:01,464 INFO  org.apache.flink.runtime.taskmanager.Task                     - Source: Custom Source -&gt; Map -&gt; Timestamps/Watermarks (1/1) (c5607a434c6da19872cf5e0a8dda761a) switched from DEPLOYING to RUNNING.
  2020-07-02 05:36:02,369 INFO  org.apache.flink.runtime.taskmanager.Task                     - Window(TumblingEventTimeWindows(5000), EventTimeTrigger, SumField, CollectionWindow) -&gt; Sink: Unnamed (1/1) (6dfa060a1e92735ba60e28c7e5871a56) switched from RUNNING to FAILED.
  java.lang.OutOfMemoryError: Metaspace
      at java.lang.ClassLoader.defineClass1(Native Method)
      at java.lang.ClassLoader.defineClass(ClassLoader.java:756)
      at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
      at java.net.URLClassLoader.defineClass(URLClassLoader.java:468)
      at java.net.URLClassLoader.access$100(URLClassLoader.java:74)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:369)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:363)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:362)
      at org.apache.flink.util.ChildFirstClassLoader.loadClass(ChildFirstClassLoader.java:66)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
      at java.lang.Class.forName0(Native Method)
      at java.lang.Class.forName(Class.java:348)
      at org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:78)
      at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1925)
      at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1808)
      at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2099)
      at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1625)
      at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2344)
      at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2268)
      at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2126)
      at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1625)
      at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2344)
      at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2268)
      at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2126)
      at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1625)
      at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2344)
      at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2268)
      at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2126)
      at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1625)
      at java.io.ObjectInputStream.readObject(ObjectInputStream.java:465)
      at java.io.ObjectInputStream.readObject(ObjectInputStream.java:423)
  </code></pre>
  <p>and this is my part of Apache Flink jobs code:</p>
  <pre><code>public static void pumpConsumeHandler() throws Exception {
          final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          FlinkUtil.initEnv(env);
          RMQSource dataSource = FlinkUtil.initDatasource(&quot;goblin.pump.report.realtime&quot;);
          DataStream&lt;String&gt; dataStreamSource = env.addSource(dataSource);
          DataStream&lt;ReportPump&gt; record =
                  dataStreamSource.map(new MapFunction&lt;String, ReportPump&gt;() {
                      @Override
                      public ReportPump map(String value) throws Exception {
                          ObjectMapper mapper = new ObjectMapper();
                          ReportPump pumpRecord = mapper.readValue(value, ReportPump.class);
                          return pumpRecord;
                      }
                  }).assignTimestampsAndWatermarks(new CustomAssign&lt;ReportPump&gt;());
  
          record.keyBy(new KeySelector&lt;ReportPump, Tuple2&lt;Integer, Long&gt;&gt;() {
              @Override
              public Tuple2&lt;Integer, Long&gt; getKey(ReportPump value) throws Exception {
                  return Tuple2.of(value.getGameType(), value.getTenantId());
              }
          })
                  .timeWindow(Time.seconds(5))
                  .reduce(new SumField(), new CollectionWindow())
                  .addSink(new SinkFunction&lt;List&lt;ReportPump&gt;&gt;() {
                      @Override
                      public void invoke(List&lt;ReportPump&gt; reportPumps, Context context) throws Exception {
                          persistCollectData(reportPumps);
                      }
                  });
          env.execute(PumpRealtimeHandler.class.getName());
      }
  
      public static void persistCollectData(List&lt;ReportPump&gt; records) throws Exception {
          PumpConsumeService service = FlinkUtil.InitRetrofit().create(PumpConsumeService.class);
          Call&lt;ResponseBody&gt; call = service.savePumpRecords(records);
          ResponseBody body = call.execute().body();
          String str = new String(body.bytes(), &quot;UTF-8&quot;);
          ObjectMapper mapper = new ObjectMapper();
          Response response = mapper.readValue(str, Response.class);
          if (!ServiceException.API_OK.equals(response.getResultCode())) {
              log.error(&quot;save error:&quot; + str);
              throw new Exception();
          } else {
              log.debug(&quot;success：&quot; + str);
          }
      }
  </code></pre>
  <p>to solve this problem, I tweak the config:</p>
  <pre><code>taskmanager.memory.jvm-metaspace.size: 256mb(from default 96M to 256M)
  taskmanager.memory.process.size: 2g(from 1G to 2G)
  </code></pre>
  <p>I have 2 flink jobs, what should I do to avoid this problem? how much metaspace size should I config?</p>