title: When Kafka Topic partition reassignment, Flink job fails continuously
body: <h1 id="env-aq27">env</h1>
  <ul>
  <li>kafka 1.0.1</li>
  <li>flink 1.7.1</li>
  </ul>
  <h1 id="trouble-m280">trouble</h1>
  <p>I use topic with 200 partitions. and flink uses this topic. <br>
  Recently, I do manual partition reassignment. <br>
  When i reassigned partitions, Flink continuosly fails with this error. <br></p>
  <p><strong>error1.</strong></p>
  <pre><code>[2021-07-28 18:21:15,926] WARN Attempting to send response via channel for which there is no open connection, connection id ..(kafka.network.Processor)
  </code></pre>
  <p><strong>error2.</strong></p>
  <pre><code>Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring 2 record(s) for -126: 30042 ms has passed since batch creation plus linger time
  </code></pre>
  <p><strong>error3.</strong></p>
  <pre><code>java.lang.Exception: Error while triggering checkpoint 656 for Source: Custom Source -&gt; Sink: ... (32/200)
      at org.apache.flink.runtime.taskmanager.Task$1.run(Task.java:1174)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: java.lang.Exception: Could not perform checkpoint 656 for operator Source: Custom Source -&gt; Sink: ... (32/200).
      at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:570)
      at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.triggerCheckpoint(SourceStreamTask.java:116)
      at org.apache.flink.runtime.taskmanager.Task$1.run(Task.java:1163)
      ... 5 more
  Caused by: java.lang.Exception: Could not complete snapshot 656 for operator Source: Custom Source -&gt; Sink: ... (32/200).
      at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:422)
      at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.checkpointStreamOperator(StreamTask.java:1113)
      at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.executeCheckpointing(StreamTask.java:1055)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState(StreamTask.java:729)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:641)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:564)
      ... 7 more
  Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring 2 record(s) for ...-86: 30049 ms has passed since batch creation plus linger time
  </code></pre>
  <p>And When i restarted failed job, this error occurs continuously.</p>
  <pre><code>ClassLoader info: URL ClassLoader:
      file: '/blobStore-29c572a3-4ed4-48a6-b604-d93b7e4a9a10/job_8bd41a7e0690e75bd61d148d89dca963/blob_p-5c10d03a5cbb09c9a9459f1bc2a70804d0b08290-26b5562cbe83b0403b06717637e7ab47' (invalid JAR: /blobStore-29c572a3-4ed4-48a6-b604-d93b7e4a9a10/job_8bd41a7e0690e75bd61d148d89dca963/blob_p-5c10d03a5cbb09c9a9459f1bc2a70804d0b08290-26b5562cbe83b0403b06717637e7ab47 (Too many open files))
  Class not resolvable through given classloader.
  </code></pre>
  <p>So I restarted all mesos and flink cluster with zookeeper clearance.<br>
  Is there any cause to look for?</p>