How to publish mqtt message to kafka in spring

Question

Develop mqtt connector for Kafka using spring.

Using the mqtt library provided by spring, messages are collected as follows.

message handler

@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
                
                if(topic.equals("myTopic")) {
                    System.out.println("Mqtt data pub");    
                }
                System.out.println(message.getPayload());

                if(topic==null) {
                    topic = "mqttdata";
                }
                String tag = "test/vib";
                String name = null;
                if(name==null) {
                    name = KafkaMessageService.MQTT_PRODUCER;
                }
                HashMap<String, Object> datalist = new HashMap<String, Object>();
                        try {
                    datalist =convertJSONstringToMap(message.getPayload().toString());
                    System.out.println(datalist.get("mac"));
                    counts = kafkaMessageService.publish(topic, name, tag, (HashMap<String,Object>[] datalist);
                } catch (Exception e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

    public static HashMap<String,Object> convertJSONstringToMap(String json) throws Exception {
        ObjectMapper mapper = new ObjectMapper();
        HashMap<String, Object> map = new HashMap<String, Object>();
        
        map = mapper.readValue(json, new TypeReference<HashMap<String, Object>>() {});
        
        return map;
}

publish method

    public int publish(String topic,String producerName,String tag,HashMap<String,Object>[] datalist) throws NotMatchedProducerException,KafkaPubFailureException{
        KafkaProducerAdaptor adaptor = searchProducerAdaptor(producerName);
        if(adaptor==null) {
            throw new NotMatchedProducerException();
        }
        
        KafkaTemplate<String,Object> kafkaTemplate = adaptor.getKafkaTemplate();
        
        LocalDateTime currentDateTime = LocalDateTime.now();
        String receivedTime = currentDateTime.toString();
        
        ObjectMapper objectMapper = new ObjectMapper();
        
        String key = adaptor.getName();
        
        int counts = 0;
        for(HashMap<String,Object> data : datalist) {
            Map<String,Object> messagePacket = new HashMap<String,Object>();
            messagePacket.put("tag", tag);
            messagePacket.put("data", data);
            messagePacket.put("receivedtime", receivedTime);
            
            try {
                kafkaTemplate.send(topic,key,objectMapper.valueToTree(messagePacket)).get();
                logger.info("Sent message : topic=["+topic+"],key=["+key+"] value=["+messagePacket+"]");
            } catch(Exception e) {
                logger.info("Unable to send message : topic=["+topic+"],key=["+key+"] message=["+messagePacket+"] / due to : "+e.getMessage());
                throw new KafkaPubFailureException(e);
            }
            counts++;
        }
        
        return counts;
    }

I don't know how to declare a hashmap <String, object> [] as an instance and how to use it.

The above source was taken from spring support as it is, and some modifications were made.