[FLINK-20643] flink-connector-elasticsearch7 causes Direct buffer memory OOM

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Not A Bug
• Affects Version/s: 1.11.1
• Component/s: Connectors / ElasticSearch
• Labels: auto-deprioritized-major
• Assignee: Unassigned
• Reporter: Echo Lee
• Created: 17/Dec/20 08:08
• Updated: 06/Jan/22 20:49
• Resolved: 31/Dec/21 02:14

Description

flink version: 1.11.1

elasticsearch version: 6.3.1

jdk version: 1.8.0_152

deployment mode: Standalone

My job run normally under flink-connector-elasticsearch6, but failed under flink-connector-elasticsearch7. It can be solved by increasing the taskmanager.memory.framework.off-heap.size or taskmanager.memory.task.off-heap.size parameter, but I don't quite understand why it costs more off-heap memory on es7 than on es6.

error message is:

2020-12-16 15:37:08,704 ERROR org.apache.http.impl.nio.client.InternalHttpAsyncClient      [] - I/O reactor terminated abnormally
org.apache.http.nio.reactor.IOReactorException: I/O dispatch worker terminated abnormally
    at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor.execute(AbstractMultiworkerIOReactor.java:359) ~[blob_p-f8aae2a0113951c2d34918dcffc906be6d5e9c33-     80068570f848a6ac286386df457c6c1c:?]
    at org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager.execute(PoolingNHttpClientConnectionManager.java:221) ~[blob_p-                                   f8aae2a0113951c2d34918dcffc906be6d5e9c33-80068570f848a6ac286386df457c6c1c:?]
    at org.apache.http.impl.nio.client.CloseableHttpAsyncClientBase$1.run(CloseableHttpAsyncClientBase.java:64) [blob_p-f8aae2a0113951c2d34918dcffc906be6d5e9c33-          80068570f848a6ac286386df457c6c1c:?]
    at java.lang.Thread.run(Thread.java:748) [?:1.8.0_181]
Caused by: java.lang.OutOfMemoryError: Direct buffer memory
    at java.nio.Bits.reserveMemory(Bits.java:694) ~[?:1.8.0_181]
    at java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:123) ~[?:1.8.0_181]
    at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:311) ~[?:1.8.0_181]
    at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:241) ~[?:1.8.0_181]
    at sun.nio.ch.IOUtil.write(IOUtil.java:58) ~[?:1.8.0_181]
    at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:471) ~[?:1.8.0_181]
    at org.apache.http.impl.nio.codecs.AbstractContentEncoder.doWriteChunk(AbstractContentEncoder.java:171) ~[blob_p-f8aae2a0113951c2d34918dcffc906be6d5e9c33-             80068570f848a6ac286386df457c6c1c:?]
    at org.apache.http.impl.nio.codecs.AbstractContentEncoder.doWriteChunk(AbstractContentEncoder.java:164) ~[blob_p-f8aae2a0113951c2d34918dcffc906be6d5e9c33-             80068570f848a6ac286386df457c6c1c:?]
    at org.apache.http.impl.nio.codecs.AbstractContentEncoder.writeToChannel(AbstractContentEncoder.java:138) ~[blob_p-f8aae2a0113951c2d34918dcffc906be6d5e9c33-           80068570f848a6ac286386df457c6c1c:?]
    at org.apache.http.impl.nio.codecs.LengthDelimitedEncoder.write(LengthDelimitedEncoder.java:130) ~[blob_p-f8aae2a0113951c2d34918dcffc906be6d5e9c33-                    80068570f848a6ac286386df457c6c1c:?]
    at org.apache.http.nio.entity.NByteArrayEntity.produceContent(NByteArrayEntity.java:136) ~[blob_p-f8aae2a0113951c2d34918dcffc906be6d5e9c33-                            80068570f848a6ac286386df457c6c1c:?]
    at org.apache.http.nio.protocol.BasicAsyncRequestProducer.produceContent(BasicAsyncRequestProducer.java:125) ~[blob_p-f8aae2a0113951c2d34918dcffc906be6d5e9c33-        80068570f848a6ac286386df457c6c1c:?]
    at org.apache.http.impl.nio.client.MainClientExec.produceContent(MainClientExec.java:262) ~[blob_p-f8aae2a0113951c2d34918dcffc906be6d5e9c33-                           80068570f848a6ac286386df457c6c1c:?]