Kafka broker memory leak triggered by many consumers

Viewed 6341

I'm building a Java 8 application that queries a Kafka topic for exactly one message. Each request creates a new Consumer object (independent from any existing Consumer objects) which polls my Kafka topic, gets one record, and the Consumer is closed. This happens ~200k times per day and each request is independent of all others, so I don't think I can reuse consumers. Basically, a user requests a message from the topic and a consumer is created for them, then closed. This happens on average ~2 times per second, but is arbitrary, so it can happen 10 times/s or 1 time/hour, there's no way to know.

After a while, the heap size on the Kafka server (not the server running the code, but the actual server running Kafka) gets huge and garbage collection can't clear it. Eventually, more CPU time is dedicated to GC than anything else, and everything crashes until I restart Kafka.

Here's an approximate version of the code that's causing the issue, with a while(true) approximating real behaviour (in production, the consumers are not created in a while loop, but are created on-demand when a user requests a message from the topic):

Properties props = new Properties();
props.put("bootstrap.servers", "SERVER_IP:9092");
props.put("session.timeout.ms", 30000);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", 1000);

while(true){
    Consumer<String, String> consumer = new KafkaConsumer<>(props);
    TopicPartition tp = new TopicPartition("TOPIC", 0);
    consumer.assign(Arrays.asList(tp));
    consumer.seekToEnd(Arrays.asList(tp));

    // I've narrowed down the memory leak to this line
    ConsumerRecords<String, String> cr = consumer.poll(1000); 
    // If I remove this line ^, the memory leak does not happen

    /* CODE TO GET ONE RECORD */

    consumer.unsubscribe();
    consumer.close();
}

Running this code on 20 JVMs leads to a memory leak in about 20min. Here's what the heap (blue) and GC pause time (green) on the Kafka server look like: KafkaMemoryLeak

Am I doing something wrong (or is there a better way to approach this), or is this a bug in Kafka when a lot of consumers are created and closed?

I'm running Kafka 0.10.2.1 on the client side and Kafka 0.10.2.0 on the server.

3 Answers
Related