Kafka Connect 堆空间不足

Question

启动 Kafka Connect (

connect-standalone

) 后，我的任务在启动后立即失败：

java.lang.OutOfMemoryError: Java heap space
    at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
    at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
    at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:93)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
    at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
    at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:232)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:180)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:193)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:248)
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1013)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:979)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.pollConsumer(WorkerSinkTask.java:316)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:222)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:170)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:142)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
    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:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

一些Kafka文档中提到了堆空间，告诉你用“默认”来尝试，只有出现问题才修改它，但没有修改堆空间的说明。

Answer 1

当您遇到 Kafka 问题时

java.lang.OutOfMemoryError: Java heap space

这并不一定意味着这是内存问题。当尝试连接到 SSL 端口时，一些 Kafka 管理工具（例如

kafka-topics.sh

）将用此掩盖真正的错误。真正的（掩盖的）错误是

SSL handshake failed

！

请参阅此问题：https://issues.apache.org/jira/browse/KAFKA-4090

解决方案是在命令中包含一个属性文件（对于

kafka-topics.sh

，这将是

--command-config

）并绝对包含此行：

security.protocol=SSL

Answer 2

您可以通过设置

KAFKA_HEAP_OPTS

环境变量来控制最大和初始堆大小。

以下示例设置起始大小为 512 MB，最大大小为 1 GB：

KAFKA_HEAP_OPTS="-Xms512m -Xmx1g" connect-standalone connect-worker.properties connect-s3-sink.properties

运行 Kafka 命令（例如

connect-standalone

）时，会调用

kafka-run-class

脚本，该脚本会 在 KAFKA_HEAP_OPTS

 环境变量中设置默认堆大小 256 MB

（如果尚未设置）。

Answer 3

今天早上我发现了这个问题的另一个原因。我也看到了同样的异常，只是我没有使用 SSL 并且我的消息非常小。我的问题是

bootstrap-servers

网址配置错误。如果将该 URL 配置为打开但不正确的服务器和端口，则可能会导致相同的异常。 Kafka 人员已经意识到这个普遍问题，并在这里跟踪它：https://cwiki.apache.org/confluence/display/KAFKA/KIP-498%3A+Add+client-side+configuration+for+maximum+响应+大小+保护+防止+OOM

Answer 4

在我的例子中，使用利用Spring Boot Kafka自动配置的Spring Boot 2.7.8应用程序（Java代码中没有配置），问题是由于未设置安全协议引起的（显然默认值是

 PLAINTEXT

）。我与

java.lang.OutOfMemoryError: Java heap space

一起遇到的其他错误是：

Stopping container due to an Error
Error while stopping the container: 
Uncaught exception in thread 'kafka-producer-network-thread | producer-':

解决方案是将以下几行添加到我的

application.properties

：

spring.kafka.consumer.security.protocol=SSL
spring.kafka.producer.security.protocol=SSL

我尝试用以下方法修复它：

spring.kafka.security.protocol=SSL 

没用。

Answer 5

即使我面临这个问题，也无法启动给定主题的生产者和消费者。还删除了所有不必要的日志文件和主题。即使这与问题无关。

更改

kafka-run-class.sh

对我来说不起作用。我更改了以下文件

kafka-console-consumer.sh

kafka-console-生产者.sh

并停止出现 OOM 错误。此后消费者和生产者都工作得很好。

之前将尺寸增加到

KAFKA_HEAP_OPTS="-Xmx1G"

512m。

Answer 6

适用于窗户

bin\windows\kafka-server-start.bat

适用于 Linux/Mac

bin\kafka-server-start.sh

更新这些值

set KAFKA_HEAP_OPTS=-Xmx2G -Xms2G