Kafka consumer returns null

       Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KryoSerializer.class.getName());
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put("fetch.message.max.bytes", "30000000");
        KafkaConsumer<String, EmailProperties> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test"));
        
        props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KryoSerializer.class.getName());
       
        try (KafkaProducer<String, EmailProperties> producer = new KafkaProducer<>(props)) {
            String topic = "test";
            String key = "mykey";
            EmailProperties ep = new EmailProperties();
            String[] ccRecipients = { "cc1@domain.com", "cc2@domain.com" };
            String[] recipients = { "r1@domain.com", "r2@domain.com" };
            String[] picturePaths = { "picture1", "picture2" };
            ep.setCcRecipients(Arrays.asList(ccRecipients));
            ep.setFile(new File("d:/file.txt"));
            ep.setMailContent("Content");
            ep.setMimeType("text/plain");
            ep.setPicturePaths(Arrays.asList(picturePaths));
            ep.setRecipients(Arrays.asList(recipients));
            ep.setSubject("subject");
            ProducerRecord<String, EmailProperties> producerRecord = new ProducerRecord<>(topic, key, ep);
            producer.send(producerRecord);
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        ConsumerRecords<String, EmailProperties> records = consumer.poll(10000);
        consumer.close();