Проблема с пространством кучи при десериализации пакета Avro в Kafka Consumer - PullRequest
Новая
       27

Проблема с пространством кучи при десериализации пакета Avro в Kafka Consumer

2 голосов
/ 03 мая 2019

Исключение пространства кучи из памяти исключение при десериализации сообщения Avro в потребителе Kafka.

Запуск потребительского кода на Java с местным производителем и потребителем kafka, и я попытался увеличить объем кучи памяти до 10 ГБ в IntelliJ, но все еще получал эту проблему.

Простой код класса потребителя 

Properties props = new Properties();

            props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                    "localhost:9092");
            props.put(ConsumerConfig.GROUP_ID_CONFIG, "test1");
             props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
            props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
           props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                    StringDeserializer.class.getName());
            props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                    AvroDeserializer.class.getName());

            KafkaConsumer<String, BookingContext> consumer = new KafkaConsumer<>(props);


                consumer.subscribe(Arrays.asList("fastlog"));
       while (true) {
                    ConsumerRecords<String, MyClass> records = consumer.poll(100);
                    for (ConsumerRecord<String, MyClass> record : records)  
        {
                        System.out.printf("----------------------" +
                                "+\noffset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

       }
    }

Вот мой класс для десериализатора, где я написал для преобразования пакета в обычный класс после процесса. Код десериализатора Avro: 

public T deserialize(String topic, byte[] data) {
        try {
          T result = null;

          if (data != null) {
            LOGGER.debug("data='{}'", DatatypeConverter.printHexBinary(data));

            DatumReader<GenericRecord> datumReader =
                new SpecificDatumReader<>(MyClass.getClassSchema());
            Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);

            result = (T) datumReader.read(null, decoder);
            LOGGER.debug("deserialized data='{}'", result);
          }
          return result;
        } catch (Exception ex) {
          throw new SerializationException(
              "Can't deserialize data '" + Arrays.toString(data) + "' from topic '" + topic + "'", ex);
        }
      }


Exception in thread "main" java.lang.OutOfMemoryError: Java heap space
        at org.apache.avro.generic.GenericData$Array.<init>(GenericData.java:245)
        at org.apache.avro.generic.GenericDatumReader.newArray(GenericDatumReader.java:391)
        at org.apache.avro.generic.GenericDatumReader.readArray(GenericDatumReader.java:257)
        at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:177)
        at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:116)
        at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:222)
        at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:175)
        at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:116)
        at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:222)
        at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:175)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:145)
        at kafka.serializer.AvroDeserializer.deserialize(AvroDeserializer.java:59)
        at kafka.serializer.AvroDeserializer.deserialize(AvroDeserializer.java:21)
        at org.apache.kafka.common.serialization.ExtendedDeserializer$Wrapper.deserialize(ExtendedDeserializer.java:65)
        at org.apache.kafka.common.serialization.ExtendedDeserializer$Wrapper.deserialize(ExtendedDeserializer.java:55)
        at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:918)
        at org.apache.kafka.clients.consumer.internals.Fetcher.access$2600(Fetcher.java:93)
        at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.fetchRecords(Fetcher.java:1095)
        at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.access$1200(Fetcher.java:944)
        at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:567)
        at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:528)
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1110)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
        at SimpleConsumer.main(SimpleConsumer.java:43)

1 Ответ

0 голосов
/ 03 мая 2019

Код, который вы опубликовали, не показывает ничего, что могло бы исчерпать память, но вы, очевидно, храните эти result возвращенные значения где-то в другом месте, а не очищаете их. Я предлагаю вам проверить, что вызывает ваш метод deserialize, и проверить, возможно, вы сохраняете все эти результаты в списке или другой структуре данных, а не очищаете их.

Другая вещь, которую вы можете сделать, это запустить JVM-профилировщик, такой как JVisualVM, и сделать дамп кучи, который покажет вам, какой тип / количество объектов засоряет вашу кучу JVM.

...