我的Kafka连接接收器堆空间不足。还有其他类似的线程：kafka connect耗尽堆空间
问题只是在默认内存设置下运行。以前，提高到2g解决了我的问题。但是，当添加一个新的接收器时，堆错误又出现了。我把xmx提高到12g，仍然得到错误。
在我的systemd服务文件中，我有：
Environment="KAFKA_HEAP_OPTS=-Xms512m -Xmx12g"即使xmx设置非常高，我仍然会得到堆错误。我也降低了我的体重 flush.size 到1000，我想这会有帮助的。仅供参考，这个连接器是针对11个不同的Kafka主题。这会带来独特的记忆需求吗？
如何进一步修复或诊断？
仅供参考，这是与Kafka0.10.2.1和汇合平台3.2.2。最近的版本有什么改进吗？
java.lang.OutOfMemoryError: Java heap space
at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
at io.confluent.connect.s3.storage.S3OutputStream.<init>(S3OutputStream.java:67)
at io.confluent.connect.s3.storage.S3Storage.create(S3Storage.java:197)
at io.confluent.connect.s3.format.avro.AvroRecordWriterProvider$1.write(AvroRecordWriterProvider.java:67)
at io.confluent.connect.s3.TopicPartitionWriter.writeRecord(TopicPartitionWriter.java:393)
at io.confluent.connect.s3.TopicPartitionWriter.write(TopicPartitionWriter.java:197)
at io.confluent.connect.s3.S3SinkTask.put(S3SinkTask.java:173)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:429)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:250)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:179)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:139)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:182)
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)
[2018-03-13 20:31:46,398] ERROR Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerSinkTask:450)
[2018-03-13 20:31:46,401] ERROR Task avro-s3-sink-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:141)
org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:451)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:250)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:179)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:139)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:182)
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)
2条答案
按热度按时间mrwjdhj31#
目前，s3连接器的内存需求取决于未完成分区的数量和
s3.part.size. 尝试将后者设置为5mb（允许的最小值）。默认值为25mb。另请阅读此处，了解有关尺寸建议的更详细说明：
https://github.com/confluentinc/kafka-connect-storage-cloud/issues/29
0yg35tkg2#
首先，我对Kafka一无所知。
但是，作为一般规则，当一个进程遇到某种容量限制时，您不能提高该限制，那么您必须以某种方式限制该进程。建议你探索一下周期性停顿的可能性。可能睡10毫秒非常100毫秒。差不多吧。
您可以尝试的另一件事是将kafka进程固定到一个特定的cpu上。这有时会产生惊人的有益效果。