title: Flink : Memory ran out exception
body: <p>I set up a Flink cluster with the following configuration:</p>
  
  <p><a href="https://i.stack.imgur.com/9uTNn.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/9uTNn.png" alt="Job Manager"></a>
  <strong>The number of task managers: 2</strong>
  <a href="https://i.stack.imgur.com/6s2zV.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/6s2zV.png" alt="Task Manager"></a></p>
  
  <p>When I run the Connected Components algorithm (<code>org.apache.flink.graph.library.ConnectedComponents</code>) on a graph with 4 million edges and 1,750,000 vertices with parallelism 2, I get  this exception:</p>
  
  <blockquote>
    <p>java.lang.RuntimeException: Error obtaining the sorted input: Thread 'SortMerger Reading Thread' terminated due to an exception: The record exceeds the maximum size of a sort buffer (current maximum: 87949312 bytes).
        at org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619)
        at org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1108)
        at org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:82)
        at org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:503)
        at org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:368)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
        at java.lang.Thread.run(Thread.java:748)
    Caused by: java.io.IOException: Thread 'SortMerger Reading Thread' terminated due to an exception: The record exceeds the maximum size of a sort buffer (current maximum: 87949312 bytes).
        at org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:800)
    Caused by: java.io.IOException: The record exceeds the maximum size of a sort buffer (current maximum: 87949312 bytes).
        at org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ReadingThread.go(UnilateralSortMerger.java:955)
        at org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:796)</p>
  </blockquote>
  
  <p>According to my configuration, I do not know why it says <code>The record exceeds the maximum size of a sort buffer (current maximum:87949312 bytes)</code> while there is enough memory segments.</p>
  
  <p>Is there anybody knows how to solve this problem?</p>