title: High memory consumption on kafka consumer
body: <p>Recently I found that kafka consumers require a lot of ram.</p>
  
  <p>For tests I've just started locally a single-threaded consumer that listens a single topic.Topic has 4 partitions. Kafka has only one broker.</p>
  
  <p>From producer I sent only 10 small messages (it was around 11:44:30 PM, see at the image I attached at the link). Since then nobody has sent any more messages to this topic.
  From then I've been seeing on the diagram with constantly growing memory consumption during the consumer polling work. Line is growing untill GC is not   called. 
  Consumer just sends poll-requests and returns nothing but require a lot of memory. </p>
  
  <p>I think it's problem.
  I tried to do some tuning, i.e. configuring some params as FETCH_MAX_BYTES_CONFIG/MAX_PARTITION_FETCH_BYTES_CONFIG/MAX_POLL_RECORDS_CONFIG but nothing actually worked out.</p>
  
  <p>SSCCE:</p>
  
  <p>KafkaConsumerConfig:</p>
  
  <pre><code>import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.common.serialization.ByteArrayDeserializer;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Configuration;
  import org.springframework.kafka.annotation.EnableKafka;
  import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
  import org.springframework.kafka.core.ConsumerFactory;
  import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  import org.springframework.kafka.listener.AbstractMessageListenerContainer;
  
  import java.util.HashMap;
  import java.util.Map;
  
  @Configuration
  @EnableKafka
  public class KafkaConsumerConfig {
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, byte[]&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, byte[]&gt; factory
                  = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConcurrency(1);
          factory.setConsumerFactory(consumerFactory());
          factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
          return factory;
      }
  
      private ConsumerFactory&lt;String, byte[]&gt; consumerFactory() {
          return new DefaultKafkaConsumerFactory&lt;&gt;(consumerProperties());
      }
  
      private Map&lt;String, Object&gt; consumerProperties() {
          final Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
          props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, "local-test-consumer-group");
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
  
          //props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 1000000); //1mb
          //props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5);
          //props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 256000); //256kb
          return props;
      }
  
  }
  </code></pre>
  
  <p>KafkaDataListener</p>
  
  <pre><code>import org.springframework.kafka.annotation.KafkaListener;
  import org.springframework.kafka.support.Acknowledgment;
  import org.springframework.stereotype.Component;
  
  import java.util.Arrays;
  
  @Component
  public class KafkaDataListener {
  
      @KafkaListener(topics = "local-test-topic", containerFactory = "kafkaListenerContainerFactory")
      public void consumeEvent(byte[] eventData, final Acknowledgment ack) {
           try {
               System.out.println("consumer received message:" + Arrays.toString(eventData));
           }finally {
               ack.acknowledge();
           }
      }
  }
  </code></pre>
  
  <p>Main App</p>
  
  <pre><code>import org.springframework.boot.SpringApplication;
  import org.springframework.boot.autoconfigure.SpringBootApplication;
  
  @SpringBootApplication
  public class App {
      public static void main(String[] args) {
          SpringApplication.run(App.class, args);
      }
  }
  </code></pre>
  
  <p>build.gradle:</p>
  
  <pre><code>group 'org.test.kafka'
  version '1.0-SNAPSHOT'
  
  apply plugin: 'java'
  
  sourceCompatibility = 1.8
  
  repositories {
      mavenLocal()
      mavenCentral()
  }
  
  dependencies {
  
      compile group: 'org.springframework.boot', name: 'spring-boot-starter', version: '1.5.4.RELEASE'
      compile group: 'org.springframework.kafka', name: 'spring-kafka', version: '1.2.2.RELEASE'
  }
  </code></pre>
  
  <p>output:</p>
  
  <pre><code>2018-02-02 23:38:51.506  INFO 14888 --- [           main] org.kafka.tests.App                      : Starting App on so-workstation with PID 14888 (/home/.../projects/custom/kafka-consumer-test/out/production/classes started by ... in /home/.../projects/custom/kafka-consumer-test)
  2018-02-02 23:38:51.508  INFO 14888 --- [           main] org.kafka.tests.App                      : No active profile set, falling back to default profiles: default
  2018-02-02 23:38:51.591  INFO 14888 --- [           main] s.c.a.AnnotationConfigApplicationContext : Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@281e3708: startup date [Fri Feb 02 23:38:51 MSK 2018]; root of context hierarchy
  2018-02-02 23:38:52.028  INFO 14888 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.kafka.annotation.KafkaBootstrapConfiguration' of type [org.springframework.kafka.annotation.KafkaBootstrapConfiguration$$EnhancerBySpringCGLIB$$2d249aa5] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
  2018-02-02 23:38:52.205  INFO 14888 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Registering beans for JMX exposure on startup
  2018-02-02 23:38:52.207  INFO 14888 --- [           main] o.s.c.support.DefaultLifecycleProcessor  : Starting beans in phase 0
  2018-02-02 23:38:52.221  INFO 14888 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
      auto.commit.interval.ms = 5000
      auto.offset.reset = latest
      bootstrap.servers = [localhost:9092]
      check.crcs = true
      client.id = 
      connections.max.idle.ms = 540000
      enable.auto.commit = false
      exclude.internal.topics = true
      fetch.max.bytes = 52428800
      fetch.max.wait.ms = 500
      fetch.min.bytes = 1
      group.id = local-test-consumer-group
      heartbeat.interval.ms = 3000
      interceptor.classes = null
      key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
      max.partition.fetch.bytes = 1048576
      max.poll.interval.ms = 300000
      max.poll.records = 500
      metadata.max.age.ms = 300000
      metric.reporters = []
      metrics.num.samples = 2
      metrics.recording.level = INFO
      metrics.sample.window.ms = 30000
      partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
      receive.buffer.bytes = 65536
      reconnect.backoff.ms = 50
      request.timeout.ms = 305000
      retry.backoff.ms = 100
      sasl.jaas.config = null
      sasl.kerberos.kinit.cmd = /usr/bin/kinit
      sasl.kerberos.min.time.before.relogin = 60000
      sasl.kerberos.service.name = null
      sasl.kerberos.ticket.renew.jitter = 0.05
      sasl.kerberos.ticket.renew.window.factor = 0.8
      sasl.mechanism = GSSAPI
      security.protocol = PLAINTEXT
      send.buffer.bytes = 131072
      session.timeout.ms = 10000
      ssl.cipher.suites = null
      ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
      ssl.endpoint.identification.algorithm = null
      ssl.key.password = null
      ssl.keymanager.algorithm = SunX509
      ssl.keystore.location = null
      ssl.keystore.password = null
      ssl.keystore.type = JKS
      ssl.protocol = TLS
      ssl.provider = null
      ssl.secure.random.implementation = null
      ssl.trustmanager.algorithm = PKIX
      ssl.truststore.location = null
      ssl.truststore.password = null
      ssl.truststore.type = JKS
      value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
  
  2018-02-02 23:38:52.225  INFO 14888 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
      auto.commit.interval.ms = 5000
      auto.offset.reset = latest
      bootstrap.servers = [localhost:9092]
      check.crcs = true
      client.id = consumer-1
      connections.max.idle.ms = 540000
      enable.auto.commit = false
      exclude.internal.topics = true
      fetch.max.bytes = 52428800
      fetch.max.wait.ms = 500
      fetch.min.bytes = 1
      group.id = local-test-consumer-group
      heartbeat.interval.ms = 3000
      interceptor.classes = null
      key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
      max.partition.fetch.bytes = 1048576
      max.poll.interval.ms = 300000
      max.poll.records = 500
      metadata.max.age.ms = 300000
      metric.reporters = []
      metrics.num.samples = 2
      metrics.recording.level = INFO
      metrics.sample.window.ms = 30000
      partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
      receive.buffer.bytes = 65536
      reconnect.backoff.ms = 50
      request.timeout.ms = 305000
      retry.backoff.ms = 100
      sasl.jaas.config = null
      sasl.kerberos.kinit.cmd = /usr/bin/kinit
      sasl.kerberos.min.time.before.relogin = 60000
      sasl.kerberos.service.name = null
      sasl.kerberos.ticket.renew.jitter = 0.05
      sasl.kerberos.ticket.renew.window.factor = 0.8
      sasl.mechanism = GSSAPI
      security.protocol = PLAINTEXT
      send.buffer.bytes = 131072
      session.timeout.ms = 10000
      ssl.cipher.suites = null
      ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
      ssl.endpoint.identification.algorithm = null
      ssl.key.password = null
      ssl.keymanager.algorithm = SunX509
      ssl.keystore.location = null
      ssl.keystore.password = null
      ssl.keystore.type = JKS
      ssl.protocol = TLS
      ssl.provider = null
      ssl.secure.random.implementation = null
      ssl.trustmanager.algorithm = PKIX
      ssl.truststore.location = null
      ssl.truststore.password = null
      ssl.truststore.type = JKS
      value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
  
  2018-02-02 23:38:52.258  INFO 14888 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 0.10.2.0
  2018-02-02 23:38:52.258  INFO 14888 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : 576d93a8dc0cf421
  2018-02-02 23:38:52.268  INFO 14888 --- [           main] org.kafka.tests.App                      : Started App in 1.056 seconds (JVM running for 1.337)
  2018-02-02 23:38:52.308  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : Discovered coordinator kafka:9092 (id: 2147483646 rack: null) for group local-test-consumer-group.
  2018-02-02 23:38:52.312  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : Revoking previously assigned partitions [] for group local-test-consumer-group
  2018-02-02 23:38:52.313  INFO 14888 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions revoked:[]
  2018-02-02 23:38:52.313  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : (Re-)joining group local-test-consumer-group
  2018-02-02 23:38:52.319  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : Successfully joined group local-test-consumer-group with generation 1
  2018-02-02 23:38:52.320  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : Setting newly assigned partitions [local-test-topic-3, local-test-topic-2, local-test-topic-1, local-test-topic-0] for group local-test-consumer-group
  2018-02-02 23:38:52.328  INFO 14888 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned:[local-test-topic-3, local-test-topic-2, local-test-topic-1, local-test-topic-0]
  consumer received message:[...]
  consumer received message:[...]
  consumer received message:[...]
  consumer received message:[...]
  consumer received message:[...]
  consumer received message:[...]
  consumer received message:[...]
  consumer received message:[...]
  consumer received message:[...]
  consumer received message:[...]
  </code></pre>
  
  <p>Does anybody know how to tune it properly?
  Or maybe on some higher version memory usage is optimized?</p>
  
  <p>Kafka server: 0.10.2.0
  Kafka client: 0.10.2.0</p>
  
  <p>See the images:</p>
  
  <p><a href="https://i.stack.imgur.com/JXP6H.png" rel="noreferrer"><img src="https://i.stack.imgur.com/JXP6H.png" alt="kafka 0.10.2 memory consumption diagram"></a>
  <a href="https://i.stack.imgur.com/SWkSN.png" rel="noreferrer"><img src="https://i.stack.imgur.com/SWkSN.png" alt="kafka 0.10.2 memory allocation per thread diagram"></a></p>
  
  <p>UPD: </p>
  
  <p>For kafka consumer 1.0.0 (spring kafka 2.1.2) memory usage diagram looks a bit better. Now the consumption line is growing not so fast as before. 
  But now RMI TCP Connection thread consumes even more memory that kafka consumer thread.</p>
  
  <p><a href="https://i.stack.imgur.com/3UBoE.png" rel="noreferrer"><img src="https://i.stack.imgur.com/3UBoE.png" alt="kafka 1.0.0 memory consumption diagram with default settings"></a>
  <a href="https://i.stack.imgur.com/FRKq2.png" rel="noreferrer"><img src="https://i.stack.imgur.com/FRKq2.png" alt="kafka 1.0.0 memory allocation per thread diagram with default settings"></a></p>
  
  <p>Moreover it seems that consumer's params are getting affects on memory usage.
  With consumer's params <code>FETCH_MAX_BYTES_CONFIG = 1mb</code> and <code>MAX_PARTITION_FETCH_BYTES_CONFIG = 256kb</code> consumption gets lower.</p>
  
  <p><a href="https://i.stack.imgur.com/rNexF.png" rel="noreferrer"><img src="https://i.stack.imgur.com/rNexF.png" alt="kafka 1.0.0 memory consumption diagram with custom settings"></a></p>