On Fri, Aug 26, 2016 at 2:28 PM, Jaikiran Pai <jai.forums2...@gmail.com>
wrote:
We have been using Kafka 0.9.0.1 (server and Java client libraries). So
far we had been using it with plaintext transport but recently have been
considering upgrading to using SSL. It mostly works except that a
mis-configured producer (and even consumer) causes a hard to relate
OutOfMemory exception and thus causing the JVM in which the client is
running, to go into a bad state. We can consistently reproduce that OOM
very easily. We decided to check if this is something that is fixed in
0.10.0.1 so upgraded one of our test systems to that version (both
server
and client libraries) but still see the same issue. Here's how it can be
easily reproduced
1. Enable SSL listener on the broker via server.properties, as per the
Kafka documentation
listeners=PLAINTEXT://:9092,SSL://:9093
ssl.keystore.location=<location-of-keystore>
ssl.keystore.password=pass
ssl.key.password=pass
ssl.truststore.location=<location-of-truststore>
ssl.truststore.password=pass
2. Start zookeeper and kafka server
3. Create a "oom-test" topic (which will be used for these tests):
kafka-topics.sh --zookeeper localhost:2181 --create --topic *oom-test*
--partitions 1 --replication-factor 1
4. Create a simple producer which sends a single message to the topic
via
Java (new producer) APIs:
public class OOMTest {
public static void main(final String[] args) throws Exception {
final Properties kafkaProducerConfigs = new Properties();
* // NOTE: Intentionally use a SSL port without specifying
security.protocol as SSL**
**kafkaProducerConfigs.setProperty(ProducerConfig.
BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");**
* kafkaProducerConfigs.setProperty(ProducerConfig.
KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
kafkaProducerConfigs.setProperty(ProducerConfig.
VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
try (KafkaProducer<String, String> producer = new
KafkaProducer<>(kafkaProducerConfigs)) {
System.out.println("Created Kafka producer");
final String topicName = "oom-test";
final String message = "Hello OOM!";
// send a message to the topic
final Future<RecordMetadata> recordMetadataFuture =
producer.send(new ProducerRecord<>(topicName, message));
final RecordMetadata sentRecordMetadata =
recordMetadataFuture.get();
System.out.println("Sent message '" + message + "' to topic
'"
+ topicName + "'");
}
System.out.println("Tests complete");
}
}
Notice that the server URL is using a SSL endpoint localhost:9093 but
isn't specifying any of the other necessary SSL configs like
security.protocol.
5. For the sake of easily reproducing this issue run this class with a
max
heap size of 256MB (-Xmx256M). Running this code throws up the following
OutOfMemoryError in one of the Sender threads:
*18:33:25,770 ERROR [KafkaThread] - Uncaught exception in
kafka-producer-network-thread | producer-1: **
**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.readFromReada
bleChannel(NetworkReceive.java:93)
at org.apache.kafka.common.network.NetworkReceive.readFrom(
NetworkReceive.java:71)
at org.apache.kafka.common.network.KafkaChannel.receive(KafkaCh
annel.java:153)
at org.apache.kafka.common.network.KafkaChannel.read(KafkaChann
el.java:134)
at org.apache.kafka.common.network.Selector.poll(Selector.java:
286)
at org.apache.kafka.clients.NetworkClient.poll(
NetworkClient.java:256)
at org.apache.kafka.clients.producer.internals.Sender.run(Sende
r.java:216)
at org.apache.kafka.clients.producer.internals.Sender.run(Sende
r.java:128)
at java.lang.Thread.run(Thread.java:745)
Note that I set it to 256MB as heap size to easily reproduce it but this
isn't specific to that size. We have been able to reproduce it at even
516MB and higher too.
This even happens with the consumer and in fact can be reproduced out of
the box with the kafka-consumer-group.sh script. All you have to do is
run
that tool as follows:
*./kafka-consumer-groups.sh --list --bootstrap-server localhost:9093
--new-consumer*
Error while executing consumer group command Java heap space
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.readFromReada
bleChannel(NetworkReceive.java:93)
at org.apache.kafka.common.network.NetworkReceive.readFrom(
NetworkReceive.java:71)
at org.apache.kafka.common.network.KafkaChannel.receive(KafkaCh
annel.java:154)
at org.apache.kafka.common.network.KafkaChannel.read(KafkaChann
el.java:135)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(
Selector.java:323)
at org.apache.kafka.common.network.Selector.poll(Selector.java:
283)
at org.apache.kafka.clients.NetworkClient.poll(
NetworkClient.java:260)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkC
lient.clientPoll(ConsumerNetworkClient.java:360)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkC
lient.poll(ConsumerNetworkClient.java:224)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkC
lient.poll(ConsumerNetworkClient.java:192)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkC
lient.poll(ConsumerNetworkClient.java:163)
at kafka.admin.AdminClient.kafka$admin$AdminClient$$send(AdminC
lient.scala:49)
at kafka.admin.AdminClient$$anonfun$sendAnyNode$1.apply(AdminCl
ient.scala:61)
at kafka.admin.AdminClient$$anonfun$sendAnyNode$1.apply(AdminCl
ient.scala:58)
at scala.collection.immutable.List.foreach(List.scala:381)
at kafka.admin.AdminClient.sendAnyNode(AdminClient.scala:58)
at kafka.admin.AdminClient.findAllBrokers(AdminClient.scala:87)
at kafka.admin.AdminClient.listAllGroups(AdminClient.scala:96)
at kafka.admin.AdminClient.listAllGroupsFlattened(AdminClient.
scala:117)
at kafka.admin.AdminClient.listAllConsumerGroupsFlattened(
AdminClient.scala:121)
at kafka.admin.ConsumerGroupCommand$KafkaConsumerGroupService.l
ist(ConsumerGroupCommand.scala:311)
at kafka.admin.ConsumerGroupCommand$.main(ConsumerGroupCommand.
scala:63)
at kafka.admin.ConsumerGroupCommand.main(ConsumerGroupCommand.
scala)
Notice that here again I'm using the new consumer and the SSL port
without
any additional SSL configs.
Once this OOM occurs, the producer is useless since it's (background)
sender thread is dead. Not just that, since we run these
producers/consumers from within our application, this OOM trips the JVM
and
our whole JVM goes into an unstable state.
Debugging shows that the NetworkReceive class in its
readFromReadableChannel method receives a value of something like
352518912
and then goes ahead to allocate a ByteBuffer of that size. This
352518912
is approximately 300 odd MB and obviously causes allocation issues. I
suspect the value being passed over the channel is incorrect.
Of course this exception is triggered by a user config error but given
that ends up in a (almost unclear) OOM and causing the JVM to go in a
bad
state, is there a way the Kafka Java library can handle this better?
Should
I file a JIRA for this? I can share a sample application on github, if
it
helps.
-Jaikiran
