昨天遇到很奇怪的一个问题, kafka-streams OOM.

报错堆栈:
log4j:WARN No appenders could be found for logger (org.apache.kafka.streams.kstream.internals.InternalStreamsBuilder).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Exception in thread "bugu_rc_streaming-fabc5587-8f91-493c-b0de-cde8b809e442-StreamThread-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.memory.MemoryPool$1.tryAllocate(MemoryPool.java:30)
  at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:112)
  at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:424)
  at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:385)
  at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:651)
  at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:572)
  at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
  at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:539)
  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:262)
  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
  at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1281)
  at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1225)
  at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1201)
  at org.apache.kafka.streams.processor.internals.StoreChangelogReader.restore(StoreChangelogReader.java:88)
  at org.apache.kafka.streams.processor.internals.TaskManager.updateNewAndRestoringTasks(TaskManager.java:328)
  at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:867)
  at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:805)
  at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:774)
不管调多大内存，一直 oom, 换了一个消费组就正常了，这是啥情况？